10178074589

Committed 31 Jul 2024 09:34AM UTC coverage: 91.068% (+0.4%) from 90.682%

Build # 10178074589

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #973 from geo-engine/remove-XGB-update-toolchain

Remove-XGB-update-toolchain

Run Details

81 of 88 new or added lines in 29 files covered. (92.05%)

456 existing lines in 119 files now uncovered.

131088 of 143945 relevant lines covered (91.07%)

53581.03 hits per line

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

96.1

/operators/src/processing/raster_type_conversion.rs

use async_trait::async_trait;
use futures::{stream::BoxStream, StreamExt, TryFutureExt, TryStreamExt};
use geoengine_datatypes::{
    primitives::{BandSelection, RasterQueryRectangle, SpatialPartition2D},
    raster::{ConvertDataType, Pixel, RasterDataType, RasterTile2D},
};
use serde::{Deserialize, Serialize};

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

#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
pub struct RasterTypeConversionParams {
    pub output_data_type: RasterDataType,
}

/// This operator converts the type of raster data into another type. This may cause precision loss as e.g. `3.1_f32` converted to `u8` will result in `3_u8`.
/// In case the value range is to small the operator will clip the values at the bounds of the data range. An example is this: The `u32` value `10000_u32` is converted to `u8`, which has a value range of 0..256. The result is `255_u8` since this is the highest value a `u8` can represent.
pub type RasterTypeConversion = Operator<RasterTypeConversionParams, SingleRasterSource>;

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

pub struct InitializedRasterTypeConversionOperator {
    name: CanonicOperatorName,
    result_descriptor: RasterResultDescriptor,
    source: Box<dyn InitializedRasterOperator>,
}

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

        let initialized_sources = self.sources.initialize_sources(path, context).await?;
        let in_desc = initialized_sources.raster.result_descriptor();

        let out_data_type = self.params.output_data_type;

        let out_desc = RasterResultDescriptor {
            spatial_reference: in_desc.spatial_reference,
            data_type: out_data_type,
            bbox: in_desc.bbox,
            time: in_desc.time,
            resolution: in_desc.resolution,
            bands: in_desc.bands.clone(),
        };

        let initialized_operator = InitializedRasterTypeConversionOperator {
            name,
            result_descriptor: out_desc,
            source: initialized_sources.raster,
        };

        Ok(initialized_operator.boxed())
    }

    span_fn!(RasterTypeConversion);
}

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

    fn query_processor(&self) -> Result<TypedRasterQueryProcessor> {
        let source = self.source.query_processor()?;
        let out_data_type = self.result_descriptor.data_type;

        let res_op = call_on_generic_raster_processor!(source, source_proc => {
            call_generic_raster_processor!(out_data_type,
                RasterTypeConversionQueryProcessor::create_boxed(source_proc)
            )
        });

        Ok(res_op)
    }

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

pub struct RasterTypeConversionQueryProcessor<
    Q: RasterQueryProcessor<RasterType = PIn>,
    PIn: Pixel,
    POut: Pixel,
> {
    query_processor: Q,
    _p_out: std::marker::PhantomData<POut>,
}

impl<Q, PIn, POut> RasterTypeConversionQueryProcessor<Q, PIn, POut>
where
    Q: 'static + RasterQueryProcessor<RasterType = PIn>,
    RasterTile2D<PIn>: ConvertDataType<RasterTile2D<POut>>,
    PIn: Pixel,
    POut: Pixel,
{
    pub fn new(query_processor: Q) -> Self {
        Self {
            query_processor,
            _p_out: std::marker::PhantomData,
        }
    }

    pub fn create_boxed(source: Q) -> Box<dyn RasterQueryProcessor<RasterType = POut>> {
        RasterTypeConversionQueryProcessor::new(source).boxed()
    }
}

#[async_trait]
impl<Q, PIn: Pixel, POut: Pixel> QueryProcessor for RasterTypeConversionQueryProcessor<Q, PIn, POut>
where
    RasterTile2D<PIn>: ConvertDataType<RasterTile2D<POut>>,
    Q: RasterQueryProcessor<RasterType = PIn>,
{
    type Output = RasterTile2D<POut>;
    type SpatialBounds = SpatialPartition2D;
    type Selection = BandSelection;
    type ResultDescription = RasterResultDescriptor;

    async fn _query<'b>(
        &'b self,
        query: RasterQueryRectangle,
        ctx: &'b dyn QueryContext,
    ) -> Result<BoxStream<'b, Result<Self::Output>>> {
        let stream = self.query_processor.raster_query(query, ctx).await?;
        let converted_stream = stream.and_then(move |tile| {
            crate::util::spawn_blocking(|| tile.convert_data_type()).map_err(Into::into)
        });

        Ok(converted_stream.boxed())
    }

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

#[cfg(test)]
mod tests {
    use geoengine_datatypes::{
        primitives::{CacheHint, Measurement, SpatialPartition2D, SpatialResolution, TimeInterval},
        raster::{
            Grid2D, GridOrEmpty2D, MaskedGrid2D, RasterDataType, TileInformation,
            TilingSpecification,
        },
        spatial_reference::SpatialReference,
        util::test::TestDefault,
    };

    use crate::{
        engine::{ChunkByteSize, MockExecutionContext, RasterBandDescriptors},
        mock::{MockRasterSource, MockRasterSourceParams},
    };

    use super::*;

    #[tokio::test]
    #[allow(clippy::float_cmp)]
    async fn test_type_conversion() {
        let grid_shape = [2, 2].into();

        let tiling_specification = TilingSpecification {
            origin_coordinate: [0.0, 0.0].into(),
            tile_size_in_pixels: grid_shape,
        };

        let raster: MaskedGrid2D<u8> = Grid2D::new(grid_shape, vec![7_u8, 7, 7, 6]).unwrap().into();

        let ctx = MockExecutionContext::new_with_tiling_spec(tiling_specification);
        let query_ctx = ctx.mock_query_context(ChunkByteSize::test_default());

        let raster_tile = RasterTile2D::new_with_tile_info(
            TimeInterval::default(),
            TileInformation {
                global_geo_transform: TestDefault::test_default(),
                global_tile_position: [0, 0].into(),
                tile_size_in_pixels: grid_shape,
            },
            0,
            raster.into(),
            CacheHint::default(),
        );

        let mrs = MockRasterSource {
            params: MockRasterSourceParams {
                data: vec![raster_tile],
                result_descriptor: RasterResultDescriptor {
                    data_type: RasterDataType::U8,
                    spatial_reference: SpatialReference::epsg_4326().into(),
                    bbox: None,
                    time: None,
                    resolution: None,
                    bands: RasterBandDescriptors::new_single_band(),
                },
            },
        }
        .boxed();

        let op = RasterTypeConversion {
            params: RasterTypeConversionParams {
                output_data_type: RasterDataType::F32,
            },
            sources: SingleRasterSource { raster: mrs },
        }
        .boxed();

        let initialized_op = op
            .initialize(WorkflowOperatorPath::initialize_root(), &ctx)
            .await
            .unwrap();

        let result_descriptor = initialized_op.result_descriptor();

        assert_eq!(result_descriptor.data_type, RasterDataType::F32);
        assert_eq!(
            result_descriptor.bands[0].measurement,
            Measurement::Unitless
        );

        let query_processor = initialized_op.query_processor().unwrap();

        let query = geoengine_datatypes::primitives::RasterQueryRectangle {
            spatial_bounds: SpatialPartition2D::new((0., 0.).into(), (2., -2.).into()).unwrap(),
            spatial_resolution: SpatialResolution::one(),
            time_interval: TimeInterval::default(),
            attributes: BandSelection::first(),
        };

        let TypedRasterQueryProcessor::F32(typed_processor) = query_processor else {
            panic!("expected TypedRasterQueryProcessor::F32");
        };

        let stream = typed_processor
            .raster_query(query, &query_ctx)
            .await
            .unwrap();

        let results = stream.collect::<Vec<Result<RasterTile2D<f32>>>>().await;

        let result_tile = results.as_slice()[0].as_ref().unwrap();

        let result_grid = result_tile.grid_array.clone();

        match result_grid {
            GridOrEmpty2D::Grid(masked_grid) => {
                assert_eq!(masked_grid.inner_grid.shape, [2, 2].into());
                assert_eq!(masked_grid.inner_grid.data, &[7., 7., 7., 6.]);
            }
            GridOrEmpty2D::Empty(_) => panic!("expected GridOrEmpty2D::Grid"),
        }
    }
}

1	use async_trait::async_trait;
2	use futures::{stream::BoxStream, StreamExt, TryFutureExt, TryStreamExt};
3	use geoengine_datatypes::{
4	primitives::{BandSelection, RasterQueryRectangle, SpatialPartition2D},
5	raster::{ConvertDataType, Pixel, RasterDataType, RasterTile2D},
6	};
7	use serde::{Deserialize, Serialize};
8
9	use crate::engine::{
10	CanonicOperatorName, ExecutionContext, InitializedRasterOperator, InitializedSources, Operator,
11	OperatorName, QueryContext, QueryProcessor, RasterOperator, RasterQueryProcessor,
12	RasterResultDescriptor, SingleRasterSource, TypedRasterQueryProcessor, WorkflowOperatorPath,
13	};
14	use crate::util::Result;
15
UNCOV 16	#[derive(Debug, Serialize, Deserialize, Clone)]	×
17	#[serde(rename_all = "camelCase")]
18	pub struct RasterTypeConversionParams {
19	pub output_data_type: RasterDataType,
20	}
21
22	/// This operator converts the type of raster data into another type. This may cause precision loss as e.g. `3.1_f32` converted to `u8` will result in `3_u8`.
23	/// In case the value range is to small the operator will clip the values at the bounds of the data range. An example is this: The `u32` value `10000_u32` is converted to `u8`, which has a value range of 0..256. The result is `255_u8` since this is the highest value a `u8` can represent.
24	pub type RasterTypeConversion = Operator<RasterTypeConversionParams, SingleRasterSource>;
25
26	impl OperatorName for RasterTypeConversion {
27	const TYPE_NAME: &'static str = "RasterTypeConversion";
28	}
29
30	pub struct InitializedRasterTypeConversionOperator {
31	name: CanonicOperatorName,
32	result_descriptor: RasterResultDescriptor,
33	source: Box<dyn InitializedRasterOperator>,
34	}
35
36	#[typetag::serde]	×
37	#[async_trait]
38	impl RasterOperator for RasterTypeConversion {
39	async fn _initialize(
40	self: Box<Self>,
41	path: WorkflowOperatorPath,
42	context: &dyn ExecutionContext,
43	) -> Result<Box<dyn InitializedRasterOperator>> {	2✔
44	let name = CanonicOperatorName::from(&self);	2✔
45		2✔
46	let initialized_sources = self.sources.initialize_sources(path, context).await?;	137✔
47	let in_desc = initialized_sources.raster.result_descriptor();	2✔
48		2✔
49	let out_data_type = self.params.output_data_type;	2✔
50		2✔
51	let out_desc = RasterResultDescriptor {	2✔
52	spatial_reference: in_desc.spatial_reference,	2✔
53	data_type: out_data_type,	2✔
54	bbox: in_desc.bbox,	2✔
55	time: in_desc.time,	2✔
56	resolution: in_desc.resolution,	2✔
57	bands: in_desc.bands.clone(),	2✔
58	};	2✔
59		2✔
60	let initialized_operator = InitializedRasterTypeConversionOperator {	2✔
61	name,	2✔
62	result_descriptor: out_desc,	2✔
63	source: initialized_sources.raster,	2✔
64	};	2✔
65		2✔
66	Ok(initialized_operator.boxed())	2✔
67	}	2✔
68
69	span_fn!(RasterTypeConversion);
70	}
71
72	impl InitializedRasterOperator for InitializedRasterTypeConversionOperator {
73	fn result_descriptor(&self) -> &RasterResultDescriptor {	2✔
74	&self.result_descriptor	2✔
75	}	2✔
76
77	fn query_processor(&self) -> Result<TypedRasterQueryProcessor> {	2✔
78	let source = self.source.query_processor()?;	2✔
79	let out_data_type = self.result_descriptor.data_type;	2✔
80
81	let res_op = call_on_generic_raster_processor!(source, source_proc => {	2✔
82	call_generic_raster_processor!(out_data_type,	2✔
UNCOV 83	RasterTypeConversionQueryProcessor::create_boxed(source_proc)	×
84	)
85	});
86
87	Ok(res_op)	2✔
88	}	2✔
89
90	fn canonic_name(&self) -> CanonicOperatorName {	×
91	self.name.clone()	×
92	}	×
93	}
94
95	pub struct RasterTypeConversionQueryProcessor<
96	Q: RasterQueryProcessor<RasterType = PIn>,
97	PIn: Pixel,
98	POut: Pixel,
99	> {
100	query_processor: Q,
101	_p_out: std::marker::PhantomData<POut>,
102	}
103
104	impl<Q, PIn, POut> RasterTypeConversionQueryProcessor<Q, PIn, POut>
105	where
106	Q: 'static + RasterQueryProcessor<RasterType = PIn>,
107	RasterTile2D<PIn>: ConvertDataType<RasterTile2D<POut>>,
108	PIn: Pixel,
109	POut: Pixel,
110	{
111	pub fn new(query_processor: Q) -> Self {	22✔
112	Self {	22✔
113	query_processor,	22✔
114	_p_out: std::marker::PhantomData,	22✔
115	}	22✔
116	}	22✔
117
118	pub fn create_boxed(source: Q) -> Box<dyn RasterQueryProcessor<RasterType = POut>> {	2✔
119	RasterTypeConversionQueryProcessor::new(source).boxed()	2✔
120	}	2✔
121	}
122
123	#[async_trait]
124	impl<Q, PIn: Pixel, POut: Pixel> QueryProcessor for RasterTypeConversionQueryProcessor<Q, PIn, POut>
125	where
126	RasterTile2D<PIn>: ConvertDataType<RasterTile2D<POut>>,
127	Q: RasterQueryProcessor<RasterType = PIn>,
128	{
129	type Output = RasterTile2D<POut>;
130	type SpatialBounds = SpatialPartition2D;
131	type Selection = BandSelection;
132	type ResultDescription = RasterResultDescriptor;
133
134	async fn _query<'b>(
135	&'b self,
136	query: RasterQueryRectangle,
137	ctx: &'b dyn QueryContext,
138	) -> Result<BoxStream<'b, Result<Self::Output>>> {	34✔
139	let stream = self.query_processor.raster_query(query, ctx).await?;	34✔
140	let converted_stream = stream.and_then(move \|tile\| {	144✔
141	crate::util::spawn_blocking(\|\| tile.convert_data_type()).map_err(Into::into)	144✔
142	});	144✔
143		34✔
144	Ok(converted_stream.boxed())	34✔
145	}	34✔
146
147	fn result_descriptor(&self) -> &Self::ResultDescription {	68✔
148	self.query_processor.raster_result_descriptor()	68✔
149	}	68✔
150	}
151
152	#[cfg(test)]
153	mod tests {
154	use geoengine_datatypes::{
155	primitives::{CacheHint, Measurement, SpatialPartition2D, SpatialResolution, TimeInterval},
156	raster::{
157	Grid2D, GridOrEmpty2D, MaskedGrid2D, RasterDataType, TileInformation,
158	TilingSpecification,
159	},
160	spatial_reference::SpatialReference,
161	util::test::TestDefault,
162	};
163
164	use crate::{
165	engine::{ChunkByteSize, MockExecutionContext, RasterBandDescriptors},
166	mock::{MockRasterSource, MockRasterSourceParams},
167	};
168
169	use super::*;
170
171	#[tokio::test]
172	#[allow(clippy::float_cmp)]
173	async fn test_type_conversion() {	1✔
174	let grid_shape = [2, 2].into();	1✔
175		1✔
176	let tiling_specification = TilingSpecification {	1✔
177	origin_coordinate: [0.0, 0.0].into(),	1✔
178	tile_size_in_pixels: grid_shape,	1✔
179	};	1✔
180		1✔
181	let raster: MaskedGrid2D<u8> = Grid2D::new(grid_shape, vec![7_u8, 7, 7, 6]).unwrap().into();	1✔
182		1✔
183	let ctx = MockExecutionContext::new_with_tiling_spec(tiling_specification);	1✔
184	let query_ctx = ctx.mock_query_context(ChunkByteSize::test_default());	1✔
185		1✔
186	let raster_tile = RasterTile2D::new_with_tile_info(	1✔
187	TimeInterval::default(),	1✔
188	TileInformation {	1✔
189	global_geo_transform: TestDefault::test_default(),	1✔
190	global_tile_position: [0, 0].into(),	1✔
191	tile_size_in_pixels: grid_shape,	1✔
192	},	1✔
193	0,	1✔
194	raster.into(),	1✔
195	CacheHint::default(),	1✔
196	);	1✔
197		1✔
198	let mrs = MockRasterSource {	1✔
199	params: MockRasterSourceParams {	1✔
200	data: vec![raster_tile],	1✔
201	result_descriptor: RasterResultDescriptor {	1✔
202	data_type: RasterDataType::U8,	1✔
203	spatial_reference: SpatialReference::epsg_4326().into(),	1✔
204	bbox: None,	1✔
205	time: None,	1✔
206	resolution: None,	1✔
207	bands: RasterBandDescriptors::new_single_band(),	1✔
208	},	1✔
209	},	1✔
210	}	1✔
211	.boxed();	1✔
212		1✔
213	let op = RasterTypeConversion {	1✔
214	params: RasterTypeConversionParams {	1✔
215	output_data_type: RasterDataType::F32,	1✔
216	},	1✔
217	sources: SingleRasterSource { raster: mrs },	1✔
218	}	1✔
219	.boxed();	1✔
220		1✔
221	let initialized_op = op	1✔
222	.initialize(WorkflowOperatorPath::initialize_root(), &ctx)	1✔
223	.await	1✔
224	.unwrap();	1✔
225		1✔
226	let result_descriptor = initialized_op.result_descriptor();	1✔
227		1✔
228	assert_eq!(result_descriptor.data_type, RasterDataType::F32);	1✔
229	assert_eq!(	1✔
230	result_descriptor.bands[0].measurement,	1✔
231	Measurement::Unitless	1✔
232	);	1✔
233		1✔
234	let query_processor = initialized_op.query_processor().unwrap();	1✔
235		1✔
236	let query = geoengine_datatypes::primitives::RasterQueryRectangle {	1✔
237	spatial_bounds: SpatialPartition2D::new((0., 0.).into(), (2., -2.).into()).unwrap(),	1✔
238	spatial_resolution: SpatialResolution::one(),	1✔
239	time_interval: TimeInterval::default(),	1✔
240	attributes: BandSelection::first(),	1✔
241	};	1✔
242		1✔
243	let TypedRasterQueryProcessor::F32(typed_processor) = query_processor else {	1✔
244	panic!("expected TypedRasterQueryProcessor::F32");	1✔
245	};	1✔
246		1✔
247	let stream = typed_processor	1✔
248	.raster_query(query, &query_ctx)	1✔
249	.await	1✔
250	.unwrap();	1✔
251		1✔
252	let results = stream.collect::<Vec<Result<RasterTile2D<f32>>>>().await;	1✔
253		1✔
254	let result_tile = results.as_slice()[0].as_ref().unwrap();	1✔
255		1✔
256	let result_grid = result_tile.grid_array.clone();	1✔
257		1✔
258	match result_grid {	1✔
259	GridOrEmpty2D::Grid(masked_grid) => {	1✔
260	assert_eq!(masked_grid.inner_grid.shape, [2, 2].into());	1✔
261	assert_eq!(masked_grid.inner_grid.data, &[7., 7., 7., 6.]);	1✔
262	}	1✔
263	GridOrEmpty2D::Empty(_) => panic!("expected GridOrEmpty2D::Grid"),	1✔
264	}	1✔
265	}	1✔
266	}

geo-engine / geoengine / 10178074589

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