18890899227

Committed 28 Oct 2025 10:21PM UTC coverage: 88.317%. First build

Build # 18890899227

Build Type

Pull #1084

github

Committed by

web-flow

Commit Message

Merge 2f2fc88dc into 85068105d

Pull Request Pull Request #1084: feat: Add time_query() and TimeDescriptor

Coverage Stats

2409 of 2689 new or added lines in 73 files covered. (89.59%)

115141 of 130373 relevant lines covered (88.32%)

225661.33 hits per line

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

92.36

/operators/src/processing/raster_scaling.rs

use crate::engine::{
    BoxRasterQueryProcessor, CanonicOperatorName, ExecutionContext, InitializedRasterOperator,
    InitializedSources, Operator, OperatorName, QueryProcessor, RasterBandDescriptor,
    RasterOperator, RasterQueryProcessor, RasterResultDescriptor, SingleRasterSource,
    TypedRasterQueryProcessor, WorkflowOperatorPath,
};
use crate::util::Result;
use async_trait::async_trait;
use futures::{StreamExt, TryStreamExt};

use geoengine_datatypes::raster::{
    CheckedMulThenAddTransformation, CheckedSubThenDivTransformation, ElementScaling,
    ScalingTransformation,
};
use geoengine_datatypes::{
    primitives::Measurement,
    raster::{Pixel, RasterPropertiesKey, RasterTile2D},
};
use num::FromPrimitive;
use num_traits::AsPrimitive;
use rayon::ThreadPool;
use serde::{Deserialize, Serialize};
use std::marker::PhantomData;
use std::sync::Arc;

#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
pub struct RasterScalingParams {
    slope: SlopeOffsetSelection,
    offset: SlopeOffsetSelection,
    output_measurement: Option<Measurement>,
    scaling_mode: ScalingMode,
}

#[derive(Debug, Serialize, Deserialize, Clone, Copy)]
#[serde(rename_all = "camelCase")]
pub enum ScalingMode {
    MulSlopeAddOffset,
    SubOffsetDivSlope,
}

#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase", tag = "type")]
enum SlopeOffsetSelection {
    Auto,
    MetadataKey(RasterPropertiesKey),
    Constant { value: f64 },
}

impl Default for SlopeOffsetSelection {
    fn default() -> Self {
        Self::Auto
    }
}

/// The raster scaling operator scales/unscales the values of a raster by a given scale factor and offset.
/// This is done by applying the following formulas to every pixel.
/// For unscaling the formula is:
///
/// `p_new = p_old * slope + offset`
///
/// For scaling the formula is:
///
/// `p_new = (p_old - offset) / slope`
///
/// `p_old` and `p_new` refer to the old and new pixel values,
/// The slope and offset values are either properties attached to the input raster or a fixed value.
///
/// An example for Meteosat Second Generation properties is:
///
/// - offset: `msg.calibration_offset`
/// - slope: `msg.calibration_slope`
pub type RasterScaling = Operator<RasterScalingParams, SingleRasterSource>;

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

pub struct InitializedRasterScalingOperator {
    name: CanonicOperatorName,
    path: WorkflowOperatorPath,
    slope: SlopeOffsetSelection,
    offset: SlopeOffsetSelection,
    result_descriptor: RasterResultDescriptor,
    source: Box<dyn InitializedRasterOperator>,
    scaling_mode: ScalingMode,
}

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

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

        let out_desc = RasterResultDescriptor {
            spatial_reference: in_desc.spatial_reference,
            data_type: in_desc.data_type,
            time: in_desc.time,
            spatial_grid: in_desc.spatial_grid,
            bands: in_desc
                .bands
                .iter()
                .map(|b| {
                    RasterBandDescriptor::new(
                        b.name.clone(),
                        self.params
                            .output_measurement
                            .clone()
                            .unwrap_or_else(|| b.measurement.clone()),
                    )
                })
                .collect::<Vec<_>>()
                .try_into()?,
        };

        let initialized_operator = InitializedRasterScalingOperator {
            name,
            path,
            slope: self.params.slope,
            offset: self.params.offset,
            result_descriptor: out_desc,
            source: input.raster,
            scaling_mode: self.params.scaling_mode,
        };

        Ok(initialized_operator.boxed())
    }

    span_fn!(RasterScaling);
}

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

    fn query_processor(&self) -> Result<TypedRasterQueryProcessor> {
        let slope = self.slope.clone();
        let offset = self.offset.clone();
        let source = self.source.query_processor()?;
        let scaling_mode = self.scaling_mode;

        let res = match scaling_mode {
            ScalingMode::SubOffsetDivSlope => {
                call_on_generic_raster_processor!(source, source_proc => { TypedRasterQueryProcessor::from(create_boxed_processor::<_,_, CheckedSubThenDivTransformation>(self.result_descriptor.clone(), slope, offset,  source_proc)) })
            }
            ScalingMode::MulSlopeAddOffset => {
                call_on_generic_raster_processor!(source, source_proc => { TypedRasterQueryProcessor::from(create_boxed_processor::<_,_, CheckedMulThenAddTransformation>(self.result_descriptor.clone(), slope, offset,  source_proc)) })
            }
        };

        Ok(res)
    }

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

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

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

struct RasterTransformationProcessor<Q, P, S>
where
    Q: RasterQueryProcessor<RasterType = P>,
{
    source: Q,
    result_descriptor: RasterResultDescriptor,
    slope: SlopeOffsetSelection,
    offset: SlopeOffsetSelection,
    _transformation: PhantomData<S>,
}

fn create_boxed_processor<Q, P, S>(
    result_descriptor: RasterResultDescriptor,
    slope: SlopeOffsetSelection,
    offset: SlopeOffsetSelection,
    source: Q,
) -> BoxRasterQueryProcessor<P>
where
    Q: RasterQueryProcessor<RasterType = P> + 'static,
    P: Pixel + FromPrimitive + 'static + Default,
    f64: AsPrimitive<P>,
    S: Send + Sync + 'static + ScalingTransformation<P>,
{
    RasterTransformationProcessor::<Q, P, S>::create(result_descriptor, slope, offset, source)
        .boxed()
}

impl<Q, P, S> RasterTransformationProcessor<Q, P, S>
where
    Q: RasterQueryProcessor<RasterType = P> + 'static,
    P: Pixel + FromPrimitive + 'static + Default,
    f64: AsPrimitive<P>,
    S: Send + Sync + 'static + ScalingTransformation<P>,
{
    pub fn create(
        result_descriptor: RasterResultDescriptor,
        slope: SlopeOffsetSelection,
        offset: SlopeOffsetSelection,
        source: Q,
    ) -> RasterTransformationProcessor<Q, P, S> {
        RasterTransformationProcessor {
            source,
            result_descriptor,
            slope,
            offset,
            _transformation: PhantomData,
        }
    }

    async fn scale_tile_async(
        &self,
        tile: RasterTile2D<P>,
        _pool: Arc<ThreadPool>,
    ) -> Result<RasterTile2D<P>> {
        // either use the provided metadata/constant or the default values from the properties
        let offset = match &self.offset {
            SlopeOffsetSelection::MetadataKey(key) => tile.properties.number_property::<P>(key)?,
            SlopeOffsetSelection::Constant { value } => value.as_(),
            SlopeOffsetSelection::Auto => tile.properties.offset().as_(),
        };

        let slope = match &self.slope {
            SlopeOffsetSelection::MetadataKey(key) => tile.properties.number_property::<P>(key)?,
            SlopeOffsetSelection::Constant { value } => value.as_(),
            SlopeOffsetSelection::Auto => tile.properties.scale().as_(),
        };

        let res_tile =
            crate::util::spawn_blocking(move || tile.transform_elements::<S>(slope, offset))
                .await?;

        Ok(res_tile)
    }
}

#[async_trait]
impl<Q, P, S> QueryProcessor for RasterTransformationProcessor<Q, P, S>
where
    P: Pixel + FromPrimitive + 'static + Default,
    f64: AsPrimitive<P>,
    Q: RasterQueryProcessor<RasterType = P> + 'static,
    S: Send + Sync + 'static + ScalingTransformation<P>,
{
    type Output = RasterTile2D<P>;
    type SpatialBounds = Q::SpatialBounds;
    type ResultDescription = RasterResultDescriptor;
    type Selection = Q::Selection;

    async fn _query<'a>(
        &'a self,
        query: geoengine_datatypes::primitives::RasterQueryRectangle,
        ctx: &'a dyn crate::engine::QueryContext,
    ) -> Result<futures::stream::BoxStream<'a, Result<Self::Output>>> {
        let src = self.source.raster_query(query, ctx).await?;
        let rs = src.and_then(move |tile| self.scale_tile_async(tile, ctx.thread_pool().clone()));
        Ok(rs.boxed())
    }

    fn result_descriptor(&self) -> &RasterResultDescriptor {
        &self.result_descriptor
    }
}

#[async_trait]
impl<Q, P, S> RasterQueryProcessor for RasterTransformationProcessor<Q, P, S>
where
    P: Pixel + FromPrimitive + 'static + Default,
    f64: AsPrimitive<P>,
    Q: RasterQueryProcessor<RasterType = P> + 'static,
    S: Send + Sync + 'static + ScalingTransformation<P>,
{
    type RasterType = P;

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

#[cfg(test)]
mod tests {

    use crate::{
        engine::{
            ChunkByteSize, MockExecutionContext, RasterBandDescriptors, SpatialGridDescriptor,
            TimeDescriptor,
        },
        mock::{MockRasterSource, MockRasterSourceParams},
    };
    use geoengine_datatypes::{
        primitives::{BandSelection, CacheHint, Coordinate2D, TimeInterval},
        raster::{
            BoundedGrid, GeoTransform, Grid2D, GridBoundingBox2D, GridOrEmpty2D, GridShape,
            GridShape2D, MaskedGrid2D, RasterDataType, RasterProperties, TileInformation,
            TilingSpecification,
        },
        spatial_reference::SpatialReference,
        util::test::TestDefault,
    };

    use super::*;

    #[tokio::test]
    async fn test_unscale() {
        let tile_size_in_pixels = GridShape2D::new_2d(2, 2);
        let result_descriptor = RasterResultDescriptor {
            data_type: RasterDataType::U8,
            spatial_reference: SpatialReference::epsg_4326().into(),
            time: TimeDescriptor::new_irregular(Some(TimeInterval::default())),
            spatial_grid: SpatialGridDescriptor::source_from_parts(
                GeoTransform::new(Coordinate2D::new(0., 0.), 1., -1.),
                tile_size_in_pixels.bounding_box(),
            ),
            bands: RasterBandDescriptors::new_single_band(),
        };

        let tiling_specification = TilingSpecification::new(tile_size_in_pixels);
        let raster =
            MaskedGrid2D::from(Grid2D::new(tile_size_in_pixels, vec![7_u8, 7, 7, 6]).unwrap());

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

        let mut raster_props = RasterProperties::default();
        raster_props.set_scale(2.0);
        raster_props.set_offset(1.0);

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

        let mrs = MockRasterSource {
            params: MockRasterSourceParams {
                data: vec![raster_tile],
                result_descriptor,
            },
        }
        .boxed();

        let scaling_mode = ScalingMode::MulSlopeAddOffset;

        let output_measurement = None;

        let op = RasterScaling {
            params: RasterScalingParams {
                slope: SlopeOffsetSelection::Auto,
                offset: SlopeOffsetSelection::Auto,
                output_measurement,
                scaling_mode,
            },
            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::U8);
        assert_eq!(
            result_descriptor.bands[0].measurement,
            Measurement::Unitless
        );

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

        let query = geoengine_datatypes::primitives::RasterQueryRectangle::new(
            GridBoundingBox2D::new([0, 0], [1, 1]).unwrap(),
            TimeInterval::default(),
            BandSelection::first(),
        );

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

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

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

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

        let result_grid = result_tile.grid_array.clone();

        match result_grid {
            GridOrEmpty2D::Grid(grid) => {
                assert_eq!(grid.shape(), &GridShape::new([2, 2]));

                let res = grid.masked_element_deref_iterator().collect::<Vec<_>>();

                let expected = vec![Some(15), Some(15), Some(15), Some(13)];

                assert_eq!(res, expected);
            }
            GridOrEmpty2D::Empty(_) => panic!("expected GridOrEmpty2D::Grid"),
        }
    }

    #[tokio::test]
    async fn test_scale() {
        let tile_size_in_pixels = GridShape2D::new_2d(2, 2);
        let result_descriptor = RasterResultDescriptor {
            data_type: RasterDataType::U8,
            spatial_reference: SpatialReference::epsg_4326().into(),
            time: TimeDescriptor::new_irregular(Some(TimeInterval::default())),
            spatial_grid: SpatialGridDescriptor::source_from_parts(
                GeoTransform::new(Coordinate2D::new(0., 0.), 1., -1.),
                tile_size_in_pixels.bounding_box(),
            ),
            bands: RasterBandDescriptors::new_single_band(),
        };

        let tiling_specification = TilingSpecification::new(tile_size_in_pixels);

        let raster =
            MaskedGrid2D::from(Grid2D::new(tile_size_in_pixels, vec![15_u8, 15, 15, 13]).unwrap());

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

        let mut raster_props = RasterProperties::default();
        raster_props.set_scale(2.0);
        raster_props.set_offset(1.0);

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

        let mrs = MockRasterSource {
            params: MockRasterSourceParams {
                data: vec![raster_tile],
                result_descriptor,
            },
        }
        .boxed();

        let scaling_mode = ScalingMode::SubOffsetDivSlope;

        let output_measurement = None;

        let params = RasterScalingParams {
            slope: SlopeOffsetSelection::Auto,
            offset: SlopeOffsetSelection::Auto,
            output_measurement,
            scaling_mode,
        };

        let op = RasterScaling {
            params,
            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::U8);
        assert_eq!(
            result_descriptor.bands[0].measurement,
            Measurement::Unitless
        );

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

        let query = geoengine_datatypes::primitives::RasterQueryRectangle::new(
            GridBoundingBox2D::new([0, 0], [1, 1]).unwrap(),
            TimeInterval::default(),
            BandSelection::first(),
        );

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

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

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

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

        let result_grid = result_tile.grid_array.clone();

        match result_grid {
            GridOrEmpty2D::Grid(grid) => {
                assert_eq!(grid.shape(), &GridShape::new([2, 2]));

                let res = grid.masked_element_deref_iterator().collect::<Vec<_>>();

                let expected = vec![Some(7), Some(7), Some(7), Some(6)];

                assert_eq!(res, expected);
            }
            GridOrEmpty2D::Empty(_) => panic!("expected GridOrEmpty2D::Grid"),
        }
    }
}

1	use crate::engine::{
2	BoxRasterQueryProcessor, CanonicOperatorName, ExecutionContext, InitializedRasterOperator,
3	InitializedSources, Operator, OperatorName, QueryProcessor, RasterBandDescriptor,
4	RasterOperator, RasterQueryProcessor, RasterResultDescriptor, SingleRasterSource,
5	TypedRasterQueryProcessor, WorkflowOperatorPath,
6	};
7	use crate::util::Result;
8	use async_trait::async_trait;
9	use futures::{StreamExt, TryStreamExt};
10
11	use geoengine_datatypes::raster::{
12	CheckedMulThenAddTransformation, CheckedSubThenDivTransformation, ElementScaling,
13	ScalingTransformation,
14	};
15	use geoengine_datatypes::{
16	primitives::Measurement,
17	raster::{Pixel, RasterPropertiesKey, RasterTile2D},
18	};
19	use num::FromPrimitive;
20	use num_traits::AsPrimitive;
21	use rayon::ThreadPool;
22	use serde::{Deserialize, Serialize};
23	use std::marker::PhantomData;
24	use std::sync::Arc;
25
26	#[derive(Debug, Serialize, Deserialize, Clone)]
27	#[serde(rename_all = "camelCase")]
28	pub struct RasterScalingParams {
29	slope: SlopeOffsetSelection,
30	offset: SlopeOffsetSelection,
31	output_measurement: Option<Measurement>,
32	scaling_mode: ScalingMode,
33	}
34
35	#[derive(Debug, Serialize, Deserialize, Clone, Copy)]
36	#[serde(rename_all = "camelCase")]
37	pub enum ScalingMode {
38	MulSlopeAddOffset,
39	SubOffsetDivSlope,
40	}
41
42	#[derive(Debug, Serialize, Deserialize, Clone)]
43	#[serde(rename_all = "camelCase", tag = "type")]
44	enum SlopeOffsetSelection {
45	Auto,
46	MetadataKey(RasterPropertiesKey),
47	Constant { value: f64 },
48	}
49
50	impl Default for SlopeOffsetSelection {
51	fn default() -> Self {	×
52	Self::Auto	×
53	}	×
54	}
55
56	/// The raster scaling operator scales/unscales the values of a raster by a given scale factor and offset.
57	/// This is done by applying the following formulas to every pixel.
58	/// For unscaling the formula is:
59	///
60	/// `p_new = p_old * slope + offset`
61	///
62	/// For scaling the formula is:
63	///
64	/// `p_new = (p_old - offset) / slope`
65	///
66	/// `p_old` and `p_new` refer to the old and new pixel values,
67	/// The slope and offset values are either properties attached to the input raster or a fixed value.
68	///
69	/// An example for Meteosat Second Generation properties is:
70	///
71	/// - offset: `msg.calibration_offset`
72	/// - slope: `msg.calibration_slope`
73	pub type RasterScaling = Operator<RasterScalingParams, SingleRasterSource>;
74
75	impl OperatorName for RasterScaling {
76	const TYPE_NAME: &'static str = "RasterScaling";
77	}
78
79	pub struct InitializedRasterScalingOperator {
80	name: CanonicOperatorName,
81	path: WorkflowOperatorPath,
82	slope: SlopeOffsetSelection,
83	offset: SlopeOffsetSelection,
84	result_descriptor: RasterResultDescriptor,
85	source: Box<dyn InitializedRasterOperator>,
86	scaling_mode: ScalingMode,
87	}
88
89	#[typetag::serde]	×
90	#[async_trait]
91	impl RasterOperator for RasterScaling {
92	async fn _initialize(
93	self: Box<Self>,
94	path: WorkflowOperatorPath,
95	context: &dyn ExecutionContext,
96	) -> Result<Box<dyn InitializedRasterOperator>> {	4✔
97	let name = CanonicOperatorName::from(&self);	2✔
98
99	let input = self	2✔
100	.sources	2✔
101	.initialize_sources(path.clone(), context)	2✔
102	.await?;	2✔
103	let in_desc = input.raster.result_descriptor();	2✔
104
105	let out_desc = RasterResultDescriptor {	2✔
106	spatial_reference: in_desc.spatial_reference,	2✔
107	data_type: in_desc.data_type,	2✔
108	time: in_desc.time,	2✔
109	spatial_grid: in_desc.spatial_grid,	2✔
110	bands: in_desc	2✔
111	.bands	2✔
112	.iter()	2✔
113	.map(\|b\| {	2✔
114	RasterBandDescriptor::new(	2✔
115	b.name.clone(),	2✔
116	self.params	2✔
117	.output_measurement	2✔
118	.clone()	2✔
119	.unwrap_or_else(\|\| b.measurement.clone()),	2✔
120	)
121	})	2✔
122	.collect::<Vec<_>>()	2✔
123	.try_into()?,	2✔
124	};
125
126	let initialized_operator = InitializedRasterScalingOperator {	2✔
127	name,	2✔
128	path,	2✔
129	slope: self.params.slope,	2✔
130	offset: self.params.offset,	2✔
131	result_descriptor: out_desc,	2✔
132	source: input.raster,	2✔
133	scaling_mode: self.params.scaling_mode,	2✔
134	};	2✔
135
136	Ok(initialized_operator.boxed())	2✔
137	}	4✔
138
139	span_fn!(RasterScaling);
140	}
141
142	impl InitializedRasterOperator for InitializedRasterScalingOperator {
143	fn result_descriptor(&self) -> &RasterResultDescriptor {	2✔
144	&self.result_descriptor	2✔
145	}	2✔
146
147	fn query_processor(&self) -> Result<TypedRasterQueryProcessor> {	2✔
148	let slope = self.slope.clone();	2✔
149	let offset = self.offset.clone();	2✔
150	let source = self.source.query_processor()?;	2✔
151	let scaling_mode = self.scaling_mode;	2✔
152
153	let res = match scaling_mode {	2✔
154	ScalingMode::SubOffsetDivSlope => {
155	call_on_generic_raster_processor!(source, source_proc => { TypedRasterQueryProcessor::from(create_boxed_processor::<_,_, CheckedSubThenDivTransformation>(self.result_descriptor.clone(), slope, offset, source_proc)) })	1✔
156	}
157	ScalingMode::MulSlopeAddOffset => {
158	call_on_generic_raster_processor!(source, source_proc => { TypedRasterQueryProcessor::from(create_boxed_processor::<_,_, CheckedMulThenAddTransformation>(self.result_descriptor.clone(), slope, offset, source_proc)) })	1✔
159	}
160	};
161
162	Ok(res)	2✔
163	}	2✔
164
165	fn canonic_name(&self) -> CanonicOperatorName {	×
166	self.name.clone()	×
167	}	×
168
169	fn name(&self) -> &'static str {	×
170	RasterScaling::TYPE_NAME	×
171	}	×
172
173	fn path(&self) -> WorkflowOperatorPath {	×
174	self.path.clone()	×
175	}	×
176	}
177
178	struct RasterTransformationProcessor<Q, P, S>
179	where
180	Q: RasterQueryProcessor<RasterType = P>,
181	{
182	source: Q,
183	result_descriptor: RasterResultDescriptor,
184	slope: SlopeOffsetSelection,
185	offset: SlopeOffsetSelection,
186	_transformation: PhantomData<S>,
187	}
188
189	fn create_boxed_processor<Q, P, S>(	2✔
190	result_descriptor: RasterResultDescriptor,	2✔
191	slope: SlopeOffsetSelection,	2✔
192	offset: SlopeOffsetSelection,	2✔
193	source: Q,	2✔
194	) -> BoxRasterQueryProcessor<P>	2✔
195	where	2✔
196	Q: RasterQueryProcessor<RasterType = P> + 'static,	2✔
197	P: Pixel + FromPrimitive + 'static + Default,	2✔
198	f64: AsPrimitive<P>,	2✔
199	S: Send + Sync + 'static + ScalingTransformation<P>,	2✔
200	{
201	RasterTransformationProcessor::<Q, P, S>::create(result_descriptor, slope, offset, source)	2✔
202	.boxed()	2✔
203	}	2✔
204
205	impl<Q, P, S> RasterTransformationProcessor<Q, P, S>
206	where
207	Q: RasterQueryProcessor<RasterType = P> + 'static,
208	P: Pixel + FromPrimitive + 'static + Default,
209	f64: AsPrimitive<P>,
210	S: Send + Sync + 'static + ScalingTransformation<P>,
211	{
212	pub fn create(	2✔
213	result_descriptor: RasterResultDescriptor,	2✔
214	slope: SlopeOffsetSelection,	2✔
215	offset: SlopeOffsetSelection,	2✔
216	source: Q,	2✔
217	) -> RasterTransformationProcessor<Q, P, S> {	2✔
218	RasterTransformationProcessor {	2✔
219	source,	2✔
220	result_descriptor,	2✔
221	slope,	2✔
222	offset,	2✔
223	_transformation: PhantomData,	2✔
224	}	2✔
225	}	2✔
226
227	async fn scale_tile_async(	2✔
228	&self,	2✔
229	tile: RasterTile2D<P>,	2✔
230	_pool: Arc<ThreadPool>,	2✔
231	) -> Result<RasterTile2D<P>> {	2✔
232	// either use the provided metadata/constant or the default values from the properties
233	let offset = match &self.offset {	2✔
234	SlopeOffsetSelection::MetadataKey(key) => tile.properties.number_property::<P>(key)?,	×
235	SlopeOffsetSelection::Constant { value } => value.as_(),	×
236	SlopeOffsetSelection::Auto => tile.properties.offset().as_(),	2✔
237	};
238
239	let slope = match &self.slope {	2✔
240	SlopeOffsetSelection::MetadataKey(key) => tile.properties.number_property::<P>(key)?,	×
241	SlopeOffsetSelection::Constant { value } => value.as_(),	×
242	SlopeOffsetSelection::Auto => tile.properties.scale().as_(),	2✔
243	};
244
245	let res_tile =	2✔
246	crate::util::spawn_blocking(move \|\| tile.transform_elements::<S>(slope, offset))	2✔
247	.await?;	2✔
248
249	Ok(res_tile)	2✔
250	}	2✔
251	}
252
253	#[async_trait]
254	impl<Q, P, S> QueryProcessor for RasterTransformationProcessor<Q, P, S>
255	where
256	P: Pixel + FromPrimitive + 'static + Default,
257	f64: AsPrimitive<P>,
258	Q: RasterQueryProcessor<RasterType = P> + 'static,
259	S: Send + Sync + 'static + ScalingTransformation<P>,
260	{
261	type Output = RasterTile2D<P>;
262	type SpatialBounds = Q::SpatialBounds;
263	type ResultDescription = RasterResultDescriptor;
264	type Selection = Q::Selection;
265
266	async fn _query<'a>(
267	&'a self,
268	query: geoengine_datatypes::primitives::RasterQueryRectangle,
269	ctx: &'a dyn crate::engine::QueryContext,
270	) -> Result<futures::stream::BoxStream<'a, Result<Self::Output>>> {	4✔
271	let src = self.source.raster_query(query, ctx).await?;	2✔
272	let rs = src.and_then(move \|tile\| self.scale_tile_async(tile, ctx.thread_pool().clone()));	2✔
273	Ok(rs.boxed())	2✔
274	}	4✔
275
276	fn result_descriptor(&self) -> &RasterResultDescriptor {	4✔
277	&self.result_descriptor	4✔
278	}	4✔
279	}
280
281	#[async_trait]
282	impl<Q, P, S> RasterQueryProcessor for RasterTransformationProcessor<Q, P, S>
283	where
284	P: Pixel + FromPrimitive + 'static + Default,
285	f64: AsPrimitive<P>,
286	Q: RasterQueryProcessor<RasterType = P> + 'static,
287	S: Send + Sync + 'static + ScalingTransformation<P>,
288	{
289	type RasterType = P;
290
291	async fn time_query<'a>(
292	&'a self,
293	query: geoengine_datatypes::primitives::TimeInterval,
294	ctx: &'a dyn crate::engine::QueryContext,
295	) -> Result<futures::stream::BoxStream<'a, Result<geoengine_datatypes::primitives::TimeInterval>>>
NEW 296	{	×
NEW 297	self.source.time_query(query, ctx).await	×
NEW 298	}	×
299	}
300
301	#[cfg(test)]
302	mod tests {
303
304	use crate::{
305	engine::{
306	ChunkByteSize, MockExecutionContext, RasterBandDescriptors, SpatialGridDescriptor,
307	TimeDescriptor,
308	},
309	mock::{MockRasterSource, MockRasterSourceParams},
310	};
311	use geoengine_datatypes::{
312	primitives::{BandSelection, CacheHint, Coordinate2D, TimeInterval},
313	raster::{
314	BoundedGrid, GeoTransform, Grid2D, GridBoundingBox2D, GridOrEmpty2D, GridShape,
315	GridShape2D, MaskedGrid2D, RasterDataType, RasterProperties, TileInformation,
316	TilingSpecification,
317	},
318	spatial_reference::SpatialReference,
319	util::test::TestDefault,
320	};
321
322	use super::*;
323
324	#[tokio::test]
325	async fn test_unscale() {	1✔
326	let tile_size_in_pixels = GridShape2D::new_2d(2, 2);	1✔
327	let result_descriptor = RasterResultDescriptor {	1✔
328	data_type: RasterDataType::U8,	1✔
329	spatial_reference: SpatialReference::epsg_4326().into(),	1✔
330	time: TimeDescriptor::new_irregular(Some(TimeInterval::default())),	1✔
331	spatial_grid: SpatialGridDescriptor::source_from_parts(	1✔
332	GeoTransform::new(Coordinate2D::new(0., 0.), 1., -1.),	1✔
333	tile_size_in_pixels.bounding_box(),	1✔
334	),	1✔
335	bands: RasterBandDescriptors::new_single_band(),	1✔
336	};	1✔
337
338	let tiling_specification = TilingSpecification::new(tile_size_in_pixels);	1✔
339	let raster =	1✔
340	MaskedGrid2D::from(Grid2D::new(tile_size_in_pixels, vec![7_u8, 7, 7, 6]).unwrap());	1✔
341
342	let ctx = MockExecutionContext::new_with_tiling_spec(tiling_specification);	1✔
343	let query_ctx = ctx.mock_query_context(ChunkByteSize::test_default());	1✔
344
345	let mut raster_props = RasterProperties::default();	1✔
346	raster_props.set_scale(2.0);	1✔
347	raster_props.set_offset(1.0);	1✔
348
349	let raster_tile = RasterTile2D::new_with_tile_info_and_properties(	1✔
350	TimeInterval::default(),	1✔
351	TileInformation {	1✔
352	global_geo_transform: TestDefault::test_default(),	1✔
353	global_tile_position: [0, 0].into(),	1✔
354	tile_size_in_pixels,	1✔
355	},	1✔
356	0,
357	raster.into(),	1✔
358	raster_props,	1✔
359	CacheHint::default(),	1✔
360	);
361
362	let mrs = MockRasterSource {	1✔
363	params: MockRasterSourceParams {	1✔
364	data: vec![raster_tile],	1✔
365	result_descriptor,	1✔
366	},	1✔
367	}	1✔
368	.boxed();	1✔
369
370	let scaling_mode = ScalingMode::MulSlopeAddOffset;	1✔
371
372	let output_measurement = None;	1✔
373
374	let op = RasterScaling {	1✔
375	params: RasterScalingParams {	1✔
376	slope: SlopeOffsetSelection::Auto,	1✔
377	offset: SlopeOffsetSelection::Auto,	1✔
378	output_measurement,	1✔
379	scaling_mode,	1✔
380	},	1✔
381	sources: SingleRasterSource { raster: mrs },	1✔
382	}	1✔
383	.boxed();	1✔
384
385	let initialized_op = op	1✔
386	.initialize(WorkflowOperatorPath::initialize_root(), &ctx)	1✔
387	.await	1✔
388	.unwrap();	1✔
389
390	let result_descriptor = initialized_op.result_descriptor();	1✔
391
392	assert_eq!(result_descriptor.data_type, RasterDataType::U8);	1✔
393	assert_eq!(	1✔
394	result_descriptor.bands[0].measurement,	1✔
395	Measurement::Unitless
396	);
397
398	let query_processor = initialized_op.query_processor().unwrap();	1✔
399
400	let query = geoengine_datatypes::primitives::RasterQueryRectangle::new(	1✔
401	GridBoundingBox2D::new([0, 0], [1, 1]).unwrap(),	1✔
402	TimeInterval::default(),	1✔
403	BandSelection::first(),	1✔
404	);
405
406	let TypedRasterQueryProcessor::U8(typed_processor) = query_processor else {	1✔
407	panic!("expected TypedRasterQueryProcessor::U8");	×
408	};
409
410	let stream = typed_processor	1✔
411	.raster_query(query, &query_ctx)	1✔
412	.await	1✔
413	.unwrap();	1✔
414
415	let results = stream.collect::<Vec<Result<RasterTile2D<u8>>>>().await;	1✔
416
417	let result_tile = results.as_slice()[0].as_ref().unwrap();	1✔
418
419	let result_grid = result_tile.grid_array.clone();	1✔
420
421	match result_grid {	1✔
422	GridOrEmpty2D::Grid(grid) => {	1✔
423	assert_eq!(grid.shape(), &GridShape::new([2, 2]));	1✔
424		1✔
425	let res = grid.masked_element_deref_iterator().collect::<Vec<_>>();	1✔
426		1✔
427	let expected = vec![Some(15), Some(15), Some(15), Some(13)];	1✔
428		1✔
429	assert_eq!(res, expected);	1✔
430	}	1✔
431	GridOrEmpty2D::Empty(_) => panic!("expected GridOrEmpty2D::Grid"),	1✔
432	}	1✔
433	}	1✔
434
435	#[tokio::test]
436	async fn test_scale() {	1✔
437	let tile_size_in_pixels = GridShape2D::new_2d(2, 2);	1✔
438	let result_descriptor = RasterResultDescriptor {	1✔
439	data_type: RasterDataType::U8,	1✔
440	spatial_reference: SpatialReference::epsg_4326().into(),	1✔
441	time: TimeDescriptor::new_irregular(Some(TimeInterval::default())),	1✔
442	spatial_grid: SpatialGridDescriptor::source_from_parts(	1✔
443	GeoTransform::new(Coordinate2D::new(0., 0.), 1., -1.),	1✔
444	tile_size_in_pixels.bounding_box(),	1✔
445	),	1✔
446	bands: RasterBandDescriptors::new_single_band(),	1✔
447	};	1✔
448
449	let tiling_specification = TilingSpecification::new(tile_size_in_pixels);	1✔
450
451	let raster =	1✔
452	MaskedGrid2D::from(Grid2D::new(tile_size_in_pixels, vec![15_u8, 15, 15, 13]).unwrap());	1✔
453
454	let ctx = MockExecutionContext::new_with_tiling_spec(tiling_specification);	1✔
455	let query_ctx = ctx.mock_query_context(ChunkByteSize::test_default());	1✔
456
457	let mut raster_props = RasterProperties::default();	1✔
458	raster_props.set_scale(2.0);	1✔
459	raster_props.set_offset(1.0);	1✔
460
461	let raster_tile = RasterTile2D::new_with_tile_info_and_properties(	1✔
462	TimeInterval::default(),	1✔
463	TileInformation {	1✔
464	global_geo_transform: TestDefault::test_default(),	1✔
465	global_tile_position: [0, 0].into(),	1✔
466	tile_size_in_pixels,	1✔
467	},	1✔
468	0,
469	raster.into(),	1✔
470	raster_props,	1✔
471	CacheHint::default(),	1✔
472	);
473
474	let mrs = MockRasterSource {	1✔
475	params: MockRasterSourceParams {	1✔
476	data: vec![raster_tile],	1✔
477	result_descriptor,	1✔
478	},	1✔
479	}	1✔
480	.boxed();	1✔
481
482	let scaling_mode = ScalingMode::SubOffsetDivSlope;	1✔
483
484	let output_measurement = None;	1✔
485
486	let params = RasterScalingParams {	1✔
487	slope: SlopeOffsetSelection::Auto,	1✔
488	offset: SlopeOffsetSelection::Auto,	1✔
489	output_measurement,	1✔
490	scaling_mode,	1✔
491	};	1✔
492
493	let op = RasterScaling {	1✔
494	params,	1✔
495	sources: SingleRasterSource { raster: mrs },	1✔
496	}	1✔
497	.boxed();	1✔
498
499	let initialized_op = op	1✔
500	.initialize(WorkflowOperatorPath::initialize_root(), &ctx)	1✔
501	.await	1✔
502	.unwrap();	1✔
503
504	let result_descriptor = initialized_op.result_descriptor();	1✔
505
506	assert_eq!(result_descriptor.data_type, RasterDataType::U8);	1✔
507	assert_eq!(	1✔
508	result_descriptor.bands[0].measurement,	1✔
509	Measurement::Unitless
510	);
511
512	let query_processor = initialized_op.query_processor().unwrap();	1✔
513
514	let query = geoengine_datatypes::primitives::RasterQueryRectangle::new(	1✔
515	GridBoundingBox2D::new([0, 0], [1, 1]).unwrap(),	1✔
516	TimeInterval::default(),	1✔
517	BandSelection::first(),	1✔
518	);
519
520	let TypedRasterQueryProcessor::U8(typed_processor) = query_processor else {	1✔
521	panic!("expected TypedRasterQueryProcessor::U8");	×
522	};
523
524	let stream = typed_processor	1✔
525	.raster_query(query, &query_ctx)	1✔
526	.await	1✔
527	.unwrap();	1✔
528
529	let results = stream.collect::<Vec<Result<RasterTile2D<u8>>>>().await;	1✔
530
531	let result_tile = results.as_slice()[0].as_ref().unwrap();	1✔
532
533	let result_grid = result_tile.grid_array.clone();	1✔
534
535	match result_grid {	1✔
536	GridOrEmpty2D::Grid(grid) => {	1✔
537	assert_eq!(grid.shape(), &GridShape::new([2, 2]));	1✔
538		1✔
539	let res = grid.masked_element_deref_iterator().collect::<Vec<_>>();	1✔
540		1✔
541	let expected = vec![Some(7), Some(7), Some(7), Some(6)];	1✔
542		1✔
543	assert_eq!(res, expected);	1✔
544	}	1✔
545	GridOrEmpty2D::Empty(_) => panic!("expected GridOrEmpty2D::Grid"),	1✔
546	}	1✔
547	}	1✔
548	}

geo-engine / geoengine / 18890899227

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