19065440894

Committed 04 Nov 2025 10:21AM UTC coverage: 88.783%. First build

Build # 19065440894

Build Type

Pull #1083

github

Committed by

web-flow

Commit Message

Merge 577088df6 into 3d9be4869

Pull Request Pull Request #1083: feat: new gdal source workflow optimization

Coverage Stats

5714 of 6540 new or added lines in 71 files covered. (87.37%)

115349 of 129923 relevant lines covered (88.78%)

499710.35 hits per line

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

87.5

/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::optimization::OptimizationError;
use crate::util::Result;
use async_trait::async_trait;
use futures::{StreamExt, TryStreamExt};

use geoengine_datatypes::primitives::SpatialResolution;
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()
    }

    fn optimize(
        &self,
        target_resolution: SpatialResolution,
    ) -> Result<Box<dyn RasterOperator>, OptimizationError> {
        Ok(RasterScaling {
            params: RasterScalingParams {
                slope: self.slope.clone(),
                offset: self.offset.clone(),
                output_measurement: Some(self.result_descriptor.bands[0].measurement.clone()),
                scaling_mode: self.scaling_mode,
            },
            sources: SingleRasterSource {
                raster: self.source.optimize(target_resolution)?,
            },
        }
        .boxed())
    }
}

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

geo-engine / geoengine / 19065440894

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