19240886216

Committed 10 Nov 2025 05:48PM UTC coverage: 88.94%. First build

Build # 19240886216

Build Type

Pull #1083

github

Committed by

web-flow

Commit Message

Merge eef5160d4 into 113de40ca

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

Coverage Stats

6215 of 7057 new or added lines in 71 files covered. (88.07%)

116335 of 130802 relevant lines covered (88.94%)

496353.73 hits per line

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

88.48

/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, Default)]
#[serde(rename_all = "camelCase", tag = "type")]
enum SlopeOffsetSelection {
    #[default]
    Auto,
    MetadataKey(RasterPropertiesKey),
    Constant {
        value: f64,
    },
}

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

geo-engine / geoengine / 19240886216

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