3929938005

Build Type

push

github

Committed by GitHub

Commit Message

Merge #713

Run Details

84930 of 96741 relevant lines covered (87.79%)

79640.1 hits per line

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

95.56

/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 tracing::{span, Level};

use crate::engine::{
    CreateSpan, ExecutionContext, InitializedRasterOperator, Operator, OperatorName, QueryContext,
    QueryProcessor, RasterOperator, RasterQueryProcessor, RasterResultDescriptor,
    SingleRasterSource, TypedRasterQueryProcessor,
};
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 {
    result_descriptor: RasterResultDescriptor,
    source: Box<dyn InitializedRasterOperator>,
}

#[typetag::serde]
#[async_trait]
impl RasterOperator for RasterTypeConversion {
    async fn _initialize(
        self: Box<Self>,
        context: &dyn ExecutionContext,
    ) -> Result<Box<dyn InitializedRasterOperator>> {
        let input = self.sources.raster.initialize(context).await?;
        let in_desc = input.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 {
            result_descriptor: out_desc,
            source: input,
        };

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

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

geo-engine / geoengine / 3929938005

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