5006008836

Build Type

push

github

Committed by GitHub

Commit Message

Merge #785 #787

Run Details

936 of 936 new or added lines in 50 files covered. (100.0%)

96010 of 107707 relevant lines covered (89.14%)

72676.46 hits per line

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

93.75

/operators/src/processing/raster_type_conversion.rs

use async_trait::async_trait;
use futures::{stream::BoxStream, StreamExt, TryFutureExt, TryStreamExt};
use geoengine_datatypes::{
    primitives::{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 {
    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,
            measurement: in_desc.measurement.clone(),
            bbox: in_desc.bbox,
            time: in_desc.time,
            resolution: in_desc.resolution,
        };

        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;

    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())
    }
}

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

    use crate::{
        engine::{ChunkByteSize, MockExecutionContext},
        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,
            },
            raster.into(),
        );

        let mrs = MockRasterSource {
            params: MockRasterSourceParams {
                data: vec![raster_tile],
                result_descriptor: RasterResultDescriptor {
                    data_type: RasterDataType::U8,
                    spatial_reference: SpatialReference::epsg_4326().into(),
                    measurement: Measurement::Unitless,
                    bbox: None,
                    time: None,
                    resolution: None,
                },
            },
        }
        .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.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(),
        };

        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::{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
16	#[derive(Debug, Serialize, Deserialize, Clone)]	1✔
17	#[serde(rename_all = "camelCase")]
18	pub struct RasterTypeConversionParams {
19	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(	1✔
40	self: Box<Self>,	1✔
41	path: WorkflowOperatorPath,	1✔
42	context: &dyn ExecutionContext,	1✔
43	) -> Result<Box<dyn InitializedRasterOperator>> {	1✔
44	let name = CanonicOperatorName::from(&self);	1✔
45
46	let initialized_sources = self.sources.initialize_sources(path, context).await?;	1✔
47	let in_desc = initialized_sources.raster.result_descriptor();	1✔
48		1✔
49	let out_data_type = self.params.output_data_type;	1✔
50		1✔
51	let out_desc = RasterResultDescriptor {	1✔
52	spatial_reference: in_desc.spatial_reference,	1✔
53	data_type: out_data_type,	1✔
54	measurement: in_desc.measurement.clone(),	1✔
55	bbox: in_desc.bbox,	1✔
56	time: in_desc.time,	1✔
57	resolution: in_desc.resolution,	1✔
58	};	1✔
59		1✔
60	let initialized_operator = InitializedRasterTypeConversionOperator {	1✔
61	name,	1✔
62	result_descriptor: out_desc,	1✔
63	source: initialized_sources.raster,	1✔
64	};	1✔
65		1✔
66	Ok(initialized_operator.boxed())	1✔
67	}	2✔
68
69	span_fn!(RasterTypeConversion);	×
70	}
71
72	impl InitializedRasterOperator for InitializedRasterTypeConversionOperator {
73	fn result_descriptor(&self) -> &RasterResultDescriptor {	1✔
74	&self.result_descriptor	1✔
75	}	1✔
76
77	fn query_processor(&self) -> Result<TypedRasterQueryProcessor> {	1✔
78	let source = self.source.query_processor()?;	1✔
79	let out_data_type = self.result_descriptor.data_type;	1✔
80
81	let res_op = call_on_generic_raster_processor!(source, source_proc => {	1✔
82	call_generic_raster_processor!(out_data_type,	1✔
83	RasterTypeConversionQueryProcessor::create_boxed(source_proc)	1✔
84	)
85	});
86
87	Ok(res_op)	1✔
88	}	1✔
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 {	31✔
112	Self {	31✔
113	query_processor,	31✔
114	_p_out: std::marker::PhantomData,	31✔
115	}	31✔
116	}	31✔
117
118	pub fn create_boxed(source: Q) -> Box<dyn RasterQueryProcessor<RasterType = POut>> {	1✔
119	RasterTypeConversionQueryProcessor::new(source).boxed()	1✔
120	}	1✔
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
132	async fn _query<'b>(	45✔
133	&'b self,	45✔
134	query: RasterQueryRectangle,	45✔
135	ctx: &'b dyn QueryContext,	45✔
136	) -> Result<BoxStream<'b, Result<Self::Output>>> {	45✔
137	let stream = self.query_processor.raster_query(query, ctx).await?;	45✔
138	let converted_stream = stream.and_then(move \|tile\| {	92✔
139	crate::util::spawn_blocking(\|\| tile.convert_data_type()).map_err(Into::into)	92✔
140	});	92✔
141		45✔
142	Ok(converted_stream.boxed())	45✔
143	}	90✔
144	}
145
146	#[cfg(test)]
147	mod tests {
148	use geoengine_datatypes::{
149	primitives::{Measurement, SpatialPartition2D, SpatialResolution, TimeInterval},
150	raster::{
151	Grid2D, GridOrEmpty2D, MaskedGrid2D, RasterDataType, TileInformation,
152	TilingSpecification,
153	},
154	spatial_reference::SpatialReference,
155	util::test::TestDefault,
156	};
157
158	use crate::{
159	engine::{ChunkByteSize, MockExecutionContext},
160	mock::{MockRasterSource, MockRasterSourceParams},
161	};
162
163	use super::*;
164
165	#[tokio::test]	1✔
166	#[allow(clippy::float_cmp)]
167	async fn test_type_conversion() {	1✔
168	let grid_shape = [2, 2].into();	1✔
169		1✔
170	let tiling_specification = TilingSpecification {	1✔
171	origin_coordinate: [0.0, 0.0].into(),	1✔
172	tile_size_in_pixels: grid_shape,	1✔
173	};	1✔
174		1✔
175	let raster: MaskedGrid2D<u8> = Grid2D::new(grid_shape, vec![7_u8, 7, 7, 6]).unwrap().into();	1✔
176		1✔
177	let ctx = MockExecutionContext::new_with_tiling_spec(tiling_specification);	1✔
178	let query_ctx = ctx.mock_query_context(ChunkByteSize::test_default());	1✔
179		1✔
180	let raster_tile = RasterTile2D::new_with_tile_info(	1✔
181	TimeInterval::default(),	1✔
182	TileInformation {	1✔
183	global_geo_transform: TestDefault::test_default(),	1✔
184	global_tile_position: [0, 0].into(),	1✔
185	tile_size_in_pixels: grid_shape,	1✔
186	},	1✔
187	raster.into(),	1✔
188	);	1✔
189		1✔
190	let mrs = MockRasterSource {	1✔
191	params: MockRasterSourceParams {	1✔
192	data: vec![raster_tile],	1✔
193	result_descriptor: RasterResultDescriptor {	1✔
194	data_type: RasterDataType::U8,	1✔
195	spatial_reference: SpatialReference::epsg_4326().into(),	1✔
196	measurement: Measurement::Unitless,	1✔
197	bbox: None,	1✔
198	time: None,	1✔
199	resolution: None,	1✔
200	},	1✔
201	},	1✔
202	}	1✔
203	.boxed();	1✔
204		1✔
205	let op = RasterTypeConversion {	1✔
206	params: RasterTypeConversionParams {	1✔
207	output_data_type: RasterDataType::F32,	1✔
208	},	1✔
209	sources: SingleRasterSource { raster: mrs },	1✔
210	}	1✔
211	.boxed();	1✔
212
213	let initialized_op = op	1✔
214	.initialize(WorkflowOperatorPath::initialize_root(), &ctx)	1✔
215	.await	×
216	.unwrap();	1✔
217		1✔
218	let result_descriptor = initialized_op.result_descriptor();	1✔
219		1✔
220	assert_eq!(result_descriptor.data_type, RasterDataType::F32);	1✔
221	assert_eq!(result_descriptor.measurement, Measurement::Unitless);	1✔
222
223	let query_processor = initialized_op.query_processor().unwrap();	1✔
224		1✔
225	let query = geoengine_datatypes::primitives::RasterQueryRectangle {	1✔
226	spatial_bounds: SpatialPartition2D::new((0., 0.).into(), (2., -2.).into()).unwrap(),	1✔
227	spatial_resolution: SpatialResolution::one(),	1✔
228	time_interval: TimeInterval::default(),	1✔
229	};	1✔
230
231	let TypedRasterQueryProcessor::F32(typed_processor) = query_processor else {	1✔
232	panic!("expected TypedRasterQueryProcessor::F32");	×
233	};
234
235	let stream = typed_processor	1✔
236	.raster_query(query, &query_ctx)	1✔
237	.await	×
238	.unwrap();	1✔
239
240	let results = stream.collect::<Vec<Result<RasterTile2D<f32>>>>().await;	1✔
241
242	let result_tile = results.as_slice()[0].as_ref().unwrap();	1✔
243		1✔
244	let result_grid = result_tile.grid_array.clone();	1✔
245		1✔
246	match result_grid {	1✔
247	GridOrEmpty2D::Grid(masked_grid) => {	1✔
248	assert_eq!(masked_grid.inner_grid.shape, [2, 2].into());	1✔
249	assert_eq!(masked_grid.inner_grid.data, &[7., 7., 7., 6.]);	1✔
250	}
251	GridOrEmpty2D::Empty(_) => panic!("expected GridOrEmpty2D::Grid"),	×
252	}
253	}
254	}

geo-engine / geoengine / 5006008836

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