3929938005

Build Type

push

github

Committed by GitHub

Commit Message

Merge #713

Run Details

84930 of 96741 relevant lines covered (87.79%)

79640.1 hits per line

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

87.65

/operators/src/processing/raster_vector_join/mod.rs

mod aggregated;
mod aggregator;
mod non_aggregated;
mod util;

use crate::engine::{
    CreateSpan, ExecutionContext, InitializedRasterOperator, InitializedVectorOperator, Operator,
    OperatorName, SingleVectorMultipleRasterSources, TypedVectorQueryProcessor, VectorColumnInfo,
    VectorOperator, VectorQueryProcessor, VectorResultDescriptor,
};
use crate::error::{self, Error};
use crate::processing::raster_vector_join::non_aggregated::RasterVectorJoinProcessor;
use crate::util::Result;

use crate::processing::raster_vector_join::aggregated::RasterVectorAggregateJoinProcessor;
use async_trait::async_trait;
use futures::future::join_all;
use geoengine_datatypes::collections::VectorDataType;
use geoengine_datatypes::primitives::FeatureDataType;
use geoengine_datatypes::raster::{Pixel, RasterDataType};
use serde::{Deserialize, Serialize};
use snafu::ensure;
use tracing::{span, Level};

use self::aggregator::{
    Aggregator, FirstValueFloatAggregator, FirstValueIntAggregator, MeanValueAggregator,
    TypedAggregator,
};

/// An operator that attaches raster values to vector data
pub type RasterVectorJoin = Operator<RasterVectorJoinParams, SingleVectorMultipleRasterSources>;

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

const MAX_NUMBER_OF_RASTER_INPUTS: usize = 8;

/// The parameter spec for `RasterVectorJoin`
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct RasterVectorJoinParams {
    /// Each name reflects the output column of the join result.
    /// For each raster input, one name must be defined.
    pub names: Vec<String>,

    /// Specifies which method is used for aggregating values for a feature
    pub feature_aggregation: FeatureAggregationMethod,

    /// Specifies which method is used for aggregating values over time
    pub temporal_aggregation: TemporalAggregationMethod,
}

/// How to aggregate the values for the geometries inside a feature e.g.
/// the mean of all the raster values corresponding to the individual
/// points inside a `MultiPoint` feature.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Copy)]
#[serde(rename_all = "camelCase")]
pub enum FeatureAggregationMethod {
    First,
    Mean,
}

/// How to aggregate the values over time
/// If there are multiple rasters valid during the validity of a feature
/// the featuer is either split into multiple (None-aggregation) or the
/// values are aggreagated
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Copy)]
#[serde(rename_all = "camelCase")]
pub enum TemporalAggregationMethod {
    None,
    First,
    Mean,
}

#[typetag::serde]
#[async_trait]
impl VectorOperator for RasterVectorJoin {
    async fn _initialize(
        mut self: Box<Self>,
        context: &dyn ExecutionContext,
    ) -> Result<Box<dyn InitializedVectorOperator>> {
        ensure!(
            (1..=MAX_NUMBER_OF_RASTER_INPUTS).contains(&self.sources.rasters.len()),
            error::InvalidNumberOfRasterInputs {
                expected: 1..MAX_NUMBER_OF_RASTER_INPUTS,
                found: self.sources.rasters.len()
            }
        );
        ensure!(
            self.sources.rasters.len() == self.params.names.len(),
            error::InvalidOperatorSpec {
                reason: "`rasters` must be of equal length as `names`"
            }
        );

        let vector_source = self.sources.vector.initialize(context).await?;
        let vector_rd = vector_source.result_descriptor();

        ensure!(
            vector_rd.data_type != VectorDataType::Data,
            error::InvalidType {
                expected: format!(
                    "{}, {} or {}",
                    VectorDataType::MultiPoint,
                    VectorDataType::MultiLineString,
                    VectorDataType::MultiPolygon
                ),
                found: VectorDataType::Data.to_string()
            },
        );

        let raster_sources = join_all(
            self.sources
                .rasters
                .into_iter()
                .map(|s| s.initialize(context)),
        )
        .await
        .into_iter()
        .collect::<Result<Vec<_>>>()?;

        let spatial_reference = vector_rd.spatial_reference;

        for other_spatial_reference in raster_sources
            .iter()
            .map(|source| source.result_descriptor().spatial_reference)
        {
            ensure!(
                spatial_reference == other_spatial_reference,
                crate::error::InvalidSpatialReference {
                    expected: spatial_reference,
                    found: other_spatial_reference,
                }
            );
        }

        let params = self.params;

        let result_descriptor = vector_rd.map_columns(|columns| {
            let mut columns = columns.clone();
            for (i, new_column_name) in params.names.iter().enumerate() {
                let feature_data_type = match params.temporal_aggregation {
                    TemporalAggregationMethod::First | TemporalAggregationMethod::None => {
                        match raster_sources[i].result_descriptor().data_type {
                            RasterDataType::U8
                            | RasterDataType::U16
                            | RasterDataType::U32
                            | RasterDataType::U64
                            | RasterDataType::I8
                            | RasterDataType::I16
                            | RasterDataType::I32
                            | RasterDataType::I64 => FeatureDataType::Int,
                            RasterDataType::F32 | RasterDataType::F64 => FeatureDataType::Float,
                        }
                    }
                    TemporalAggregationMethod::Mean => FeatureDataType::Float,
                };
                columns.insert(
                    new_column_name.clone(),
                    VectorColumnInfo {
                        data_type: feature_data_type,
                        measurement: raster_sources[i].result_descriptor().measurement.clone(),
                    },
                );
            }
            columns
        });

        Ok(InitializedRasterVectorJoin {
            result_descriptor,
            vector_source,
            raster_sources,
            state: params,
        }
        .boxed())
    }

    span_fn!(RasterVectorJoin);
}

pub struct InitializedRasterVectorJoin {
    result_descriptor: VectorResultDescriptor,
    vector_source: Box<dyn InitializedVectorOperator>,
    raster_sources: Vec<Box<dyn InitializedRasterOperator>>,
    state: RasterVectorJoinParams,
}

impl InitializedVectorOperator for InitializedRasterVectorJoin {
    fn result_descriptor(&self) -> &VectorResultDescriptor {
        &self.result_descriptor
    }

    fn query_processor(&self) -> Result<TypedVectorQueryProcessor> {
        let typed_raster_processors = self
            .raster_sources
            .iter()
            .map(InitializedRasterOperator::query_processor)
            .collect::<Result<Vec<_>>>()?;

        Ok(match self.vector_source.query_processor()? {
            TypedVectorQueryProcessor::Data(_) => unreachable!(),
            TypedVectorQueryProcessor::MultiPoint(points) => {
                TypedVectorQueryProcessor::MultiPoint(match self.state.temporal_aggregation {
                    TemporalAggregationMethod::None => RasterVectorJoinProcessor::new(
                        points,
                        typed_raster_processors,
                        self.state.names.clone(),
                        self.state.feature_aggregation,
                    )
                    .boxed(),
                    TemporalAggregationMethod::First | TemporalAggregationMethod::Mean => {
                        RasterVectorAggregateJoinProcessor::new(
                            points,
                            typed_raster_processors,
                            self.state.names.clone(),
                            self.state.feature_aggregation,
                            self.state.temporal_aggregation,
                        )
                        .boxed()
                    }
                })
            }
            TypedVectorQueryProcessor::MultiPolygon(polygons) => {
                TypedVectorQueryProcessor::MultiPolygon(match self.state.temporal_aggregation {
                    TemporalAggregationMethod::None => RasterVectorJoinProcessor::new(
                        polygons,
                        typed_raster_processors,
                        self.state.names.clone(),
                        self.state.feature_aggregation,
                    )
                    .boxed(),
                    TemporalAggregationMethod::First | TemporalAggregationMethod::Mean => {
                        RasterVectorAggregateJoinProcessor::new(
                            polygons,
                            typed_raster_processors,
                            self.state.names.clone(),
                            self.state.feature_aggregation,
                            self.state.temporal_aggregation,
                        )
                        .boxed()
                    }
                })
            }
            TypedVectorQueryProcessor::MultiLineString(_) => return Err(Error::NotYetImplemented),
        })
    }
}

pub fn create_feature_aggregator<P: Pixel>(
    number_of_features: usize,
    aggregation: FeatureAggregationMethod,
) -> TypedAggregator {
    match aggregation {
        FeatureAggregationMethod::First => match P::TYPE {
            RasterDataType::U8
            | RasterDataType::U16
            | RasterDataType::U32
            | RasterDataType::U64
            | RasterDataType::I8
            | RasterDataType::I16
            | RasterDataType::I32
            | RasterDataType::I64 => FirstValueIntAggregator::new(number_of_features).into_typed(),
            RasterDataType::F32 | RasterDataType::F64 => {
                FirstValueFloatAggregator::new(number_of_features).into_typed()
            }
        },
        FeatureAggregationMethod::Mean => MeanValueAggregator::new(number_of_features).into_typed(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::str::FromStr;

    use crate::engine::{
        ChunkByteSize, MockExecutionContext, MockQueryContext, QueryProcessor, RasterOperator,
    };
    use crate::mock::MockFeatureCollectionSource;
    use crate::source::{GdalSource, GdalSourceParameters};
    use crate::util::gdal::add_ndvi_dataset;
    use futures::StreamExt;
    use geoengine_datatypes::collections::{FeatureCollectionInfos, MultiPointCollection};
    use geoengine_datatypes::dataset::DataId;
    use geoengine_datatypes::primitives::{
        BoundingBox2D, DataRef, DateTime, FeatureDataRef, MultiPoint, SpatialResolution,
        TimeInterval, VectorQueryRectangle,
    };
    use geoengine_datatypes::spatial_reference::SpatialReference;
    use geoengine_datatypes::util::{gdal::hide_gdal_errors, test::TestDefault};
    use serde_json::json;

    #[test]
    fn serialization() {
        let raster_vector_join = RasterVectorJoin {
            params: RasterVectorJoinParams {
                names: ["foo", "bar"].iter().copied().map(str::to_string).collect(),
                feature_aggregation: FeatureAggregationMethod::First,
                temporal_aggregation: TemporalAggregationMethod::Mean,
            },
            sources: SingleVectorMultipleRasterSources {
                vector: MockFeatureCollectionSource::<MultiPoint>::multiple(vec![]).boxed(),
                rasters: vec![],
            },
        };

        let serialized = json!({
            "type": "RasterVectorJoin",
            "params": {
                "names": ["foo", "bar"],
                "featureAggregation": "first",
                "temporalAggregation": "mean",
            },
            "sources": {
                "vector": {
                    "type": "MockFeatureCollectionSourceMultiPoint",
                    "params": {
                        "collections": [],
                        "spatialReference": "EPSG:4326",
                        "measurements": {},
                    }
                },
                "rasters": [],
            },
        })
        .to_string();

        let deserialized: RasterVectorJoin = serde_json::from_str(&serialized).unwrap();

        assert_eq!(deserialized.params, raster_vector_join.params);
    }

    fn ndvi_source(id: DataId) -> Box<dyn RasterOperator> {
        let gdal_source = GdalSource {
            params: GdalSourceParameters { data: id },
        };

        gdal_source.boxed()
    }

    #[tokio::test]
    async fn ndvi_time_point() {
        let point_source = MockFeatureCollectionSource::single(
            MultiPointCollection::from_data(
                MultiPoint::many(vec![
                    (-13.95, 20.05),
                    (-14.05, 20.05),
                    (-13.95, 19.95),
                    (-14.05, 19.95),
                ])
                .unwrap(),
                vec![
                    TimeInterval::new(
                        DateTime::new_utc(2014, 1, 1, 0, 0, 0),
                        DateTime::new_utc(2014, 1, 1, 0, 0, 0),
                    )
                    .unwrap();
                    4
                ],
                Default::default(),
            )
            .unwrap(),
        )
        .boxed();

        let mut exe_ctc = MockExecutionContext::test_default();
        let ndvi_id = add_ndvi_dataset(&mut exe_ctc);

        let operator = RasterVectorJoin {
            params: RasterVectorJoinParams {
                names: vec!["ndvi".to_string()],
                feature_aggregation: FeatureAggregationMethod::First,
                temporal_aggregation: TemporalAggregationMethod::First,
            },
            sources: SingleVectorMultipleRasterSources {
                vector: point_source,
                rasters: vec![ndvi_source(ndvi_id.clone())],
            },
        };

        let operator = operator.boxed().initialize(&exe_ctc).await.unwrap();

        let query_processor = operator.query_processor().unwrap().multi_point().unwrap();

        let result = query_processor
            .query(
                VectorQueryRectangle {
                    spatial_bounds: BoundingBox2D::new((-180., -90.).into(), (180., 90.).into())
                        .unwrap(),
                    time_interval: TimeInterval::default(),
                    spatial_resolution: SpatialResolution::new(0.1, 0.1).unwrap(),
                },
                &MockQueryContext::new(ChunkByteSize::MIN),
            )
            .await
            .unwrap()
            .map(Result::unwrap)
            .collect::<Vec<MultiPointCollection>>()
            .await;

        assert_eq!(result.len(), 1);

        let FeatureDataRef::Int(data) = result[0].data("ndvi").unwrap() else { unreachable!(); };

        // these values are taken from loading the tiff in QGIS
        assert_eq!(data.as_ref(), &[54, 55, 51, 55]);
    }

    #[tokio::test]
    #[allow(clippy::float_cmp)]
    async fn ndvi_time_range() {
        let point_source = MockFeatureCollectionSource::single(
            MultiPointCollection::from_data(
                MultiPoint::many(vec![
                    (-13.95, 20.05),
                    (-14.05, 20.05),
                    (-13.95, 19.95),
                    (-14.05, 19.95),
                ])
                .unwrap(),
                vec![
                    TimeInterval::new(
                        DateTime::new_utc(2014, 1, 1, 0, 0, 0),
                        DateTime::new_utc(2014, 3, 1, 0, 0, 0),
                    )
                    .unwrap();
                    4
                ],
                Default::default(),
            )
            .unwrap(),
        )
        .boxed();

        let mut exe_ctc = MockExecutionContext::test_default();
        let ndvi_id = add_ndvi_dataset(&mut exe_ctc);

        let operator = RasterVectorJoin {
            params: RasterVectorJoinParams {
                names: vec!["ndvi".to_string()],
                feature_aggregation: FeatureAggregationMethod::First,
                temporal_aggregation: TemporalAggregationMethod::Mean,
            },
            sources: SingleVectorMultipleRasterSources {
                vector: point_source,
                rasters: vec![ndvi_source(ndvi_id.clone())],
            },
        };

        let operator = operator.boxed().initialize(&exe_ctc).await.unwrap();

        let query_processor = operator.query_processor().unwrap().multi_point().unwrap();

        let result = query_processor
            .query(
                VectorQueryRectangle {
                    spatial_bounds: BoundingBox2D::new((-180., -90.).into(), (180., 90.).into())
                        .unwrap(),
                    time_interval: TimeInterval::default(),
                    spatial_resolution: SpatialResolution::new(0.1, 0.1).unwrap(),
                },
                &MockQueryContext::new(ChunkByteSize::MIN),
            )
            .await
            .unwrap()
            .map(Result::unwrap)
            .collect::<Vec<MultiPointCollection>>()
            .await;

        assert_eq!(result.len(), 1);

        let FeatureDataRef::Float(data) = result[0].data("ndvi").unwrap() else { unreachable!(); };

        // these values are taken from loading the tiff in QGIS
        assert_eq!(
            data.as_ref(),
            &[
                (54. + 52.) / 2.,
                (55. + 55.) / 2.,
                (51. + 50.) / 2.,
                (55. + 53.) / 2.,
            ]
        );
    }

    #[tokio::test]
    #[allow(clippy::float_cmp)]
    async fn ndvi_with_default_time() {
        hide_gdal_errors();

        let point_source = MockFeatureCollectionSource::single(
            MultiPointCollection::from_data(
                MultiPoint::many(vec![
                    (-13.95, 20.05),
                    (-14.05, 20.05),
                    (-13.95, 19.95),
                    (-14.05, 19.95),
                ])
                .unwrap(),
                vec![TimeInterval::default(); 4],
                Default::default(),
            )
            .unwrap(),
        )
        .boxed();

        let mut exe_ctc = MockExecutionContext::test_default();
        let ndvi_id = add_ndvi_dataset(&mut exe_ctc);

        let operator = RasterVectorJoin {
            params: RasterVectorJoinParams {
                names: vec!["ndvi".to_string()],
                feature_aggregation: FeatureAggregationMethod::First,
                temporal_aggregation: TemporalAggregationMethod::Mean,
            },
            sources: SingleVectorMultipleRasterSources {
                vector: point_source,
                rasters: vec![ndvi_source(ndvi_id.clone())],
            },
        };

        let operator = operator.boxed().initialize(&exe_ctc).await.unwrap();

        let query_processor = operator.query_processor().unwrap().multi_point().unwrap();

        let result = query_processor
            .query(
                VectorQueryRectangle {
                    spatial_bounds: BoundingBox2D::new((-180., -90.).into(), (180., 90.).into())
                        .unwrap(),
                    time_interval: TimeInterval::default(),
                    spatial_resolution: SpatialResolution::new(0.1, 0.1).unwrap(),
                },
                &MockQueryContext::new(ChunkByteSize::MIN),
            )
            .await
            .unwrap()
            .map(Result::unwrap)
            .collect::<Vec<MultiPointCollection>>()
            .await;

        assert_eq!(result.len(), 1);

        let FeatureDataRef::Float(data) = result[0].data("ndvi").unwrap() else { unreachable!(); };

        assert_eq!(data.as_ref(), &[0., 0., 0., 0.]);

        assert_eq!(data.nulls(), vec![true, true, true, true]);
    }

    #[tokio::test]
    async fn it_checks_sref() {
        let point_source = MockFeatureCollectionSource::with_collections_and_sref(
            vec![MultiPointCollection::from_data(
                MultiPoint::many(vec![
                    (-13.95, 20.05),
                    (-14.05, 20.05),
                    (-13.95, 19.95),
                    (-14.05, 19.95),
                ])
                .unwrap(),
                vec![TimeInterval::default(); 4],
                Default::default(),
            )
            .unwrap()],
            SpatialReference::from_str("EPSG:3857").unwrap(),
        )
        .boxed();

        let mut exe_ctc = MockExecutionContext::test_default();
        let ndvi_id = add_ndvi_dataset(&mut exe_ctc);

        let operator = RasterVectorJoin {
            params: RasterVectorJoinParams {
                names: vec!["ndvi".to_string()],
                feature_aggregation: FeatureAggregationMethod::First,
                temporal_aggregation: TemporalAggregationMethod::Mean,
            },
            sources: SingleVectorMultipleRasterSources {
                vector: point_source,
                rasters: vec![ndvi_source(ndvi_id.clone())],
            },
        }
        .boxed()
        .initialize(&exe_ctc)
        .await;

        assert!(matches!(
            operator,
            Err(Error::InvalidSpatialReference {
                expected: _,
                found: _,
            })
        ));
    }
}

1	mod aggregated;
2	mod aggregator;
3	mod non_aggregated;
4	mod util;
5
6	use crate::engine::{
7	CreateSpan, ExecutionContext, InitializedRasterOperator, InitializedVectorOperator, Operator,
8	OperatorName, SingleVectorMultipleRasterSources, TypedVectorQueryProcessor, VectorColumnInfo,
9	VectorOperator, VectorQueryProcessor, VectorResultDescriptor,
10	};
11	use crate::error::{self, Error};
12	use crate::processing::raster_vector_join::non_aggregated::RasterVectorJoinProcessor;
13	use crate::util::Result;
14
15	use crate::processing::raster_vector_join::aggregated::RasterVectorAggregateJoinProcessor;
16	use async_trait::async_trait;
17	use futures::future::join_all;
18	use geoengine_datatypes::collections::VectorDataType;
19	use geoengine_datatypes::primitives::FeatureDataType;
20	use geoengine_datatypes::raster::{Pixel, RasterDataType};
21	use serde::{Deserialize, Serialize};
22	use snafu::ensure;
23	use tracing::{span, Level};
24
25	use self::aggregator::{
26	Aggregator, FirstValueFloatAggregator, FirstValueIntAggregator, MeanValueAggregator,
27	TypedAggregator,
28	};
29
30	/// An operator that attaches raster values to vector data
31	pub type RasterVectorJoin = Operator<RasterVectorJoinParams, SingleVectorMultipleRasterSources>;
32
33	impl OperatorName for RasterVectorJoin {
34	const TYPE_NAME: &'static str = "RasterVectorJoin";
35	}
36
37	const MAX_NUMBER_OF_RASTER_INPUTS: usize = 8;
38
39	/// The parameter spec for `RasterVectorJoin`
40	#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]	14✔
41	#[serde(rename_all = "camelCase")]
42	pub struct RasterVectorJoinParams {
43	/// Each name reflects the output column of the join result.
44	/// For each raster input, one name must be defined.
45	pub names: Vec<String>,
46
47	/// Specifies which method is used for aggregating values for a feature
48	pub feature_aggregation: FeatureAggregationMethod,
49
50	/// Specifies which method is used for aggregating values over time
51	pub temporal_aggregation: TemporalAggregationMethod,
52	}
53
54	/// How to aggregate the values for the geometries inside a feature e.g.
55	/// the mean of all the raster values corresponding to the individual
56	/// points inside a `MultiPoint` feature.
57	#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Copy)]	4✔
58	#[serde(rename_all = "camelCase")]
59	pub enum FeatureAggregationMethod {
60	First,
61	Mean,
62	}
63
64	/// How to aggregate the values over time
65	/// If there are multiple rasters valid during the validity of a feature
66	/// the featuer is either split into multiple (None-aggregation) or the
67	/// values are aggreagated
68	#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Copy)]	4✔
69	#[serde(rename_all = "camelCase")]
70	pub enum TemporalAggregationMethod {
71	None,
72	First,
73	Mean,
74	}
75
76	#[typetag::serde]	1✔
77	#[async_trait]
78	impl VectorOperator for RasterVectorJoin {
79	async fn _initialize(	5✔
80	mut self: Box<Self>,	5✔
81	context: &dyn ExecutionContext,	5✔
82	) -> Result<Box<dyn InitializedVectorOperator>> {	5✔
83	ensure!(	5✔
84	(1..=MAX_NUMBER_OF_RASTER_INPUTS).contains(&self.sources.rasters.len()),	5✔
85	error::InvalidNumberOfRasterInputs {	×
86	expected: 1..MAX_NUMBER_OF_RASTER_INPUTS,	×
87	found: self.sources.rasters.len()	×
88	}	×
89	);
90	ensure!(	5✔
91	self.sources.rasters.len() == self.params.names.len(),	5✔
92	error::InvalidOperatorSpec {	×
93	reason: "`rasters` must be of equal length as `names`"	×
94	}	×
95	);
96
97	let vector_source = self.sources.vector.initialize(context).await?;	5✔
98	let vector_rd = vector_source.result_descriptor();	5✔
99		5✔
100	ensure!(	5✔
101	vector_rd.data_type != VectorDataType::Data,	5✔
102	error::InvalidType {	×
103	expected: format!(	×
104	"{}, {} or {}",	×
105	VectorDataType::MultiPoint,	×
106	VectorDataType::MultiLineString,	×
107	VectorDataType::MultiPolygon	×
108	),	×
109	found: VectorDataType::Data.to_string()	×
110	},	×
111	);
112
113	let raster_sources = join_all(	5✔
114	self.sources	5✔
115	.rasters	5✔
116	.into_iter()	5✔
117	.map(\|s\| s.initialize(context)),	5✔
118	)	5✔
119	.await	×
120	.into_iter()	5✔
121	.collect::<Result<Vec<_>>>()?;	5✔
122
123	let spatial_reference = vector_rd.spatial_reference;	5✔
124
125	for other_spatial_reference in raster_sources	5✔
126	.iter()	5✔
127	.map(\|source\| source.result_descriptor().spatial_reference)	5✔
128	{
129	ensure!(	5✔
130	spatial_reference == other_spatial_reference,	5✔
131	crate::error::InvalidSpatialReference {	1✔
132	expected: spatial_reference,	1✔
133	found: other_spatial_reference,	1✔
134	}	1✔
135	);
136	}
137
138	let params = self.params;	4✔
139		4✔
140	let result_descriptor = vector_rd.map_columns(\|columns\| {	4✔
141	let mut columns = columns.clone();	4✔
142	for (i, new_column_name) in params.names.iter().enumerate() {	4✔
143	let feature_data_type = match params.temporal_aggregation {	4✔
144	TemporalAggregationMethod::First \| TemporalAggregationMethod::None => {
145	match raster_sources[i].result_descriptor().data_type {	2✔
146	RasterDataType::U8
147	\| RasterDataType::U16
148	\| RasterDataType::U32
149	\| RasterDataType::U64
150	\| RasterDataType::I8
151	\| RasterDataType::I16
152	\| RasterDataType::I32
153	\| RasterDataType::I64 => FeatureDataType::Int,	2✔
154	RasterDataType::F32 \| RasterDataType::F64 => FeatureDataType::Float,	×
155	}
156	}
157	TemporalAggregationMethod::Mean => FeatureDataType::Float,	2✔
158	};
159	columns.insert(	4✔
160	new_column_name.clone(),	4✔
161	VectorColumnInfo {	4✔
162	data_type: feature_data_type,	4✔
163	measurement: raster_sources[i].result_descriptor().measurement.clone(),	4✔
164	},	4✔
165	);	4✔
166	}
167	columns	4✔
168	});	4✔
169		4✔
170	Ok(InitializedRasterVectorJoin {	4✔
171	result_descriptor,	4✔
172	vector_source,	4✔
173	raster_sources,	4✔
174	state: params,	4✔
175	}	4✔
176	.boxed())	4✔
177	}	10✔
178
179	span_fn!(RasterVectorJoin);	×
180	}
181
182	pub struct InitializedRasterVectorJoin {
183	result_descriptor: VectorResultDescriptor,
184	vector_source: Box<dyn InitializedVectorOperator>,
185	raster_sources: Vec<Box<dyn InitializedRasterOperator>>,
186	state: RasterVectorJoinParams,
187	}
188
189	impl InitializedVectorOperator for InitializedRasterVectorJoin {
190	fn result_descriptor(&self) -> &VectorResultDescriptor {	1✔
191	&self.result_descriptor	1✔
192	}	1✔
193
194	fn query_processor(&self) -> Result<TypedVectorQueryProcessor> {	4✔
195	let typed_raster_processors = self	4✔
196	.raster_sources	4✔
197	.iter()	4✔
198	.map(InitializedRasterOperator::query_processor)	4✔
199	.collect::<Result<Vec<_>>>()?;	4✔
200
201	Ok(match self.vector_source.query_processor()? {	4✔
202	TypedVectorQueryProcessor::Data(_) => unreachable!(),	×
203	TypedVectorQueryProcessor::MultiPoint(points) => {	4✔
204	TypedVectorQueryProcessor::MultiPoint(match self.state.temporal_aggregation {	4✔
205	TemporalAggregationMethod::None => RasterVectorJoinProcessor::new(	×
206	points,	×
207	typed_raster_processors,	×
208	self.state.names.clone(),	×
209	self.state.feature_aggregation,	×
210	)	×
211	.boxed(),	×
212	TemporalAggregationMethod::First \| TemporalAggregationMethod::Mean => {
213	RasterVectorAggregateJoinProcessor::new(	4✔
214	points,	4✔
215	typed_raster_processors,	4✔
216	self.state.names.clone(),	4✔
217	self.state.feature_aggregation,	4✔
218	self.state.temporal_aggregation,	4✔
219	)	4✔
220	.boxed()	4✔
221	}
222	})
223	}
224	TypedVectorQueryProcessor::MultiPolygon(polygons) => {	×
225	TypedVectorQueryProcessor::MultiPolygon(match self.state.temporal_aggregation {	×
226	TemporalAggregationMethod::None => RasterVectorJoinProcessor::new(	×
227	polygons,	×
228	typed_raster_processors,	×
229	self.state.names.clone(),	×
230	self.state.feature_aggregation,	×
231	)	×
232	.boxed(),	×
233	TemporalAggregationMethod::First \| TemporalAggregationMethod::Mean => {
234	RasterVectorAggregateJoinProcessor::new(	×
235	polygons,	×
236	typed_raster_processors,	×
237	self.state.names.clone(),	×
238	self.state.feature_aggregation,	×
239	self.state.temporal_aggregation,	×
240	)	×
241	.boxed()	×
242	}
243	})
244	}
245	TypedVectorQueryProcessor::MultiLineString(_) => return Err(Error::NotYetImplemented),	×
246	})
247	}	4✔
248	}
249
250	pub fn create_feature_aggregator<P: Pixel>(	23✔
251	number_of_features: usize,	23✔
252	aggregation: FeatureAggregationMethod,	23✔
253	) -> TypedAggregator {	23✔
254	match aggregation {	23✔
255	FeatureAggregationMethod::First => match P::TYPE {	15✔
256	RasterDataType::U8
257	\| RasterDataType::U16
258	\| RasterDataType::U32
259	\| RasterDataType::U64
260	\| RasterDataType::I8
261	\| RasterDataType::I16
262	\| RasterDataType::I32
263	\| RasterDataType::I64 => FirstValueIntAggregator::new(number_of_features).into_typed(),	15✔
264	RasterDataType::F32 \| RasterDataType::F64 => {
265	FirstValueFloatAggregator::new(number_of_features).into_typed()	×
266	}
267	},
268	FeatureAggregationMethod::Mean => MeanValueAggregator::new(number_of_features).into_typed(),	8✔
269	}
270	}	23✔
271
272	#[cfg(test)]
273	mod tests {
274	use super::*;
275	use std::str::FromStr;
276
277	use crate::engine::{
278	ChunkByteSize, MockExecutionContext, MockQueryContext, QueryProcessor, RasterOperator,
279	};
280	use crate::mock::MockFeatureCollectionSource;
281	use crate::source::{GdalSource, GdalSourceParameters};
282	use crate::util::gdal::add_ndvi_dataset;
283	use futures::StreamExt;
284	use geoengine_datatypes::collections::{FeatureCollectionInfos, MultiPointCollection};
285	use geoengine_datatypes::dataset::DataId;
286	use geoengine_datatypes::primitives::{
287	BoundingBox2D, DataRef, DateTime, FeatureDataRef, MultiPoint, SpatialResolution,
288	TimeInterval, VectorQueryRectangle,
289	};
290	use geoengine_datatypes::spatial_reference::SpatialReference;
291	use geoengine_datatypes::util::{gdal::hide_gdal_errors, test::TestDefault};
292	use serde_json::json;
293
294	#[test]	1✔
295	fn serialization() {	1✔
296	let raster_vector_join = RasterVectorJoin {	1✔
297	params: RasterVectorJoinParams {	1✔
298	names: ["foo", "bar"].iter().copied().map(str::to_string).collect(),	1✔
299	feature_aggregation: FeatureAggregationMethod::First,	1✔
300	temporal_aggregation: TemporalAggregationMethod::Mean,	1✔
301	},	1✔
302	sources: SingleVectorMultipleRasterSources {	1✔
303	vector: MockFeatureCollectionSource::<MultiPoint>::multiple(vec![]).boxed(),	1✔
304	rasters: vec![],	1✔
305	},	1✔
306	};	1✔
307		1✔
308	let serialized = json!({	1✔
309	"type": "RasterVectorJoin",	1✔
310	"params": {	1✔
311	"names": ["foo", "bar"],	1✔
312	"featureAggregation": "first",	1✔
313	"temporalAggregation": "mean",	1✔
314	},	1✔
315	"sources": {	1✔
316	"vector": {	1✔
317	"type": "MockFeatureCollectionSourceMultiPoint",	1✔
318	"params": {	1✔
319	"collections": [],	1✔
320	"spatialReference": "EPSG:4326",	1✔
321	"measurements": {},	1✔
322	}	1✔
323	},	1✔
324	"rasters": [],	1✔
325	},	1✔
326	})	1✔
327	.to_string();	1✔
328		1✔
329	let deserialized: RasterVectorJoin = serde_json::from_str(&serialized).unwrap();	1✔
330		1✔
331	assert_eq!(deserialized.params, raster_vector_join.params);	1✔
332	}	1✔
333
334	fn ndvi_source(id: DataId) -> Box<dyn RasterOperator> {	4✔
335	let gdal_source = GdalSource {	4✔
336	params: GdalSourceParameters { data: id },	4✔
337	};	4✔
338		4✔
339	gdal_source.boxed()	4✔
340	}	4✔
341
342	#[tokio::test]	1✔
343	async fn ndvi_time_point() {	1✔
344	let point_source = MockFeatureCollectionSource::single(	1✔
345	MultiPointCollection::from_data(	1✔
346	MultiPoint::many(vec![	1✔
347	(-13.95, 20.05),	1✔
348	(-14.05, 20.05),	1✔
349	(-13.95, 19.95),	1✔
350	(-14.05, 19.95),	1✔
351	])	1✔
352	.unwrap(),	1✔
353	vec![	1✔
354	TimeInterval::new(	1✔
355	DateTime::new_utc(2014, 1, 1, 0, 0, 0),	1✔
356	DateTime::new_utc(2014, 1, 1, 0, 0, 0),	1✔
357	)	1✔
358	.unwrap();	1✔
359	4	1✔
360	],	1✔
361	Default::default(),	1✔
362	)	1✔
363	.unwrap(),	1✔
364	)	1✔
365	.boxed();	1✔
366		1✔
367	let mut exe_ctc = MockExecutionContext::test_default();	1✔
368	let ndvi_id = add_ndvi_dataset(&mut exe_ctc);	1✔
369		1✔
370	let operator = RasterVectorJoin {	1✔
371	params: RasterVectorJoinParams {	1✔
372	names: vec!["ndvi".to_string()],	1✔
373	feature_aggregation: FeatureAggregationMethod::First,	1✔
374	temporal_aggregation: TemporalAggregationMethod::First,	1✔
375	},	1✔
376	sources: SingleVectorMultipleRasterSources {	1✔
377	vector: point_source,	1✔
378	rasters: vec![ndvi_source(ndvi_id.clone())],	1✔
379	},	1✔
380	};	1✔
381
382	let operator = operator.boxed().initialize(&exe_ctc).await.unwrap();	1✔
383		1✔
384	let query_processor = operator.query_processor().unwrap().multi_point().unwrap();	1✔
385
386	let result = query_processor	1✔
387	.query(	1✔
388	VectorQueryRectangle {	1✔
389	spatial_bounds: BoundingBox2D::new((-180., -90.).into(), (180., 90.).into())	1✔
390	.unwrap(),	1✔
391	time_interval: TimeInterval::default(),	1✔
392	spatial_resolution: SpatialResolution::new(0.1, 0.1).unwrap(),	1✔
393	},	1✔
394	&MockQueryContext::new(ChunkByteSize::MIN),	1✔
395	)	1✔
396	.await	×
397	.unwrap()	1✔
398	.map(Result::unwrap)	1✔
399	.collect::<Vec<MultiPointCollection>>()	1✔
400	.await;	32✔
401
402	assert_eq!(result.len(), 1);	1✔
403
404	let FeatureDataRef::Int(data) = result[0].data("ndvi").unwrap() else { unreachable!(); };	1✔
405
406	// these values are taken from loading the tiff in QGIS
407	assert_eq!(data.as_ref(), &[54, 55, 51, 55]);	1✔
408	}
409
410	#[tokio::test]	1✔
411	#[allow(clippy::float_cmp)]
412	async fn ndvi_time_range() {	1✔
413	let point_source = MockFeatureCollectionSource::single(	1✔
414	MultiPointCollection::from_data(	1✔
415	MultiPoint::many(vec![	1✔
416	(-13.95, 20.05),	1✔
417	(-14.05, 20.05),	1✔
418	(-13.95, 19.95),	1✔
419	(-14.05, 19.95),	1✔
420	])	1✔
421	.unwrap(),	1✔
422	vec![	1✔
423	TimeInterval::new(	1✔
424	DateTime::new_utc(2014, 1, 1, 0, 0, 0),	1✔
425	DateTime::new_utc(2014, 3, 1, 0, 0, 0),	1✔
426	)	1✔
427	.unwrap();	1✔
428	4	1✔
429	],	1✔
430	Default::default(),	1✔
431	)	1✔
432	.unwrap(),	1✔
433	)	1✔
434	.boxed();	1✔
435		1✔
436	let mut exe_ctc = MockExecutionContext::test_default();	1✔
437	let ndvi_id = add_ndvi_dataset(&mut exe_ctc);	1✔
438		1✔
439	let operator = RasterVectorJoin {	1✔
440	params: RasterVectorJoinParams {	1✔
441	names: vec!["ndvi".to_string()],	1✔
442	feature_aggregation: FeatureAggregationMethod::First,	1✔
443	temporal_aggregation: TemporalAggregationMethod::Mean,	1✔
444	},	1✔
445	sources: SingleVectorMultipleRasterSources {	1✔
446	vector: point_source,	1✔
447	rasters: vec![ndvi_source(ndvi_id.clone())],	1✔
448	},	1✔
449	};	1✔
450
451	let operator = operator.boxed().initialize(&exe_ctc).await.unwrap();	1✔
452		1✔
453	let query_processor = operator.query_processor().unwrap().multi_point().unwrap();	1✔
454
455	let result = query_processor	1✔
456	.query(	1✔
457	VectorQueryRectangle {	1✔
458	spatial_bounds: BoundingBox2D::new((-180., -90.).into(), (180., 90.).into())	1✔
459	.unwrap(),	1✔
460	time_interval: TimeInterval::default(),	1✔
461	spatial_resolution: SpatialResolution::new(0.1, 0.1).unwrap(),	1✔
462	},	1✔
463	&MockQueryContext::new(ChunkByteSize::MIN),	1✔
464	)	1✔
465	.await	×
466	.unwrap()	1✔
467	.map(Result::unwrap)	1✔
468	.collect::<Vec<MultiPointCollection>>()	1✔
469	.await;	65✔
470
471	assert_eq!(result.len(), 1);	1✔
472
473	let FeatureDataRef::Float(data) = result[0].data("ndvi").unwrap() else { unreachable!(); };	1✔
474
475	// these values are taken from loading the tiff in QGIS
476	assert_eq!(	1✔
477	data.as_ref(),	1✔
478	&[	1✔
479	(54. + 52.) / 2.,	1✔
480	(55. + 55.) / 2.,	1✔
481	(51. + 50.) / 2.,	1✔
482	(55. + 53.) / 2.,	1✔
483	]	1✔
484	);	1✔
485	}
486
487	#[tokio::test]	1✔
488	#[allow(clippy::float_cmp)]
489	async fn ndvi_with_default_time() {	1✔
490	hide_gdal_errors();	1✔
491		1✔
492	let point_source = MockFeatureCollectionSource::single(	1✔
493	MultiPointCollection::from_data(	1✔
494	MultiPoint::many(vec![	1✔
495	(-13.95, 20.05),	1✔
496	(-14.05, 20.05),	1✔
497	(-13.95, 19.95),	1✔
498	(-14.05, 19.95),	1✔
499	])	1✔
500	.unwrap(),	1✔
501	vec![TimeInterval::default(); 4],	1✔
502	Default::default(),	1✔
503	)	1✔
504	.unwrap(),	1✔
505	)	1✔
506	.boxed();	1✔
507		1✔
508	let mut exe_ctc = MockExecutionContext::test_default();	1✔
509	let ndvi_id = add_ndvi_dataset(&mut exe_ctc);	1✔
510		1✔
511	let operator = RasterVectorJoin {	1✔
512	params: RasterVectorJoinParams {	1✔
513	names: vec!["ndvi".to_string()],	1✔
514	feature_aggregation: FeatureAggregationMethod::First,	1✔
515	temporal_aggregation: TemporalAggregationMethod::Mean,	1✔
516	},	1✔
517	sources: SingleVectorMultipleRasterSources {	1✔
518	vector: point_source,	1✔
519	rasters: vec![ndvi_source(ndvi_id.clone())],	1✔
520	},	1✔
521	};	1✔
522
523	let operator = operator.boxed().initialize(&exe_ctc).await.unwrap();	1✔
524		1✔
525	let query_processor = operator.query_processor().unwrap().multi_point().unwrap();	1✔
526
527	let result = query_processor	1✔
528	.query(	1✔
529	VectorQueryRectangle {	1✔
530	spatial_bounds: BoundingBox2D::new((-180., -90.).into(), (180., 90.).into())	1✔
531	.unwrap(),	1✔
532	time_interval: TimeInterval::default(),	1✔
533	spatial_resolution: SpatialResolution::new(0.1, 0.1).unwrap(),	1✔
534	},	1✔
535	&MockQueryContext::new(ChunkByteSize::MIN),	1✔
536	)	1✔
537	.await	×
538	.unwrap()	1✔
539	.map(Result::unwrap)	1✔
540	.collect::<Vec<MultiPointCollection>>()	1✔
541	.await;	3✔
542
543	assert_eq!(result.len(), 1);	1✔
544
545	let FeatureDataRef::Float(data) = result[0].data("ndvi").unwrap() else { unreachable!(); };	1✔
546
547	assert_eq!(data.as_ref(), &[0., 0., 0., 0.]);	1✔
548
549	assert_eq!(data.nulls(), vec![true, true, true, true]);	1✔
550	}
551
552	#[tokio::test]	1✔
553	async fn it_checks_sref() {	1✔
554	let point_source = MockFeatureCollectionSource::with_collections_and_sref(	1✔
555	vec![MultiPointCollection::from_data(	1✔
556	MultiPoint::many(vec![	1✔
557	(-13.95, 20.05),	1✔
558	(-14.05, 20.05),	1✔
559	(-13.95, 19.95),	1✔
560	(-14.05, 19.95),	1✔
561	])	1✔
562	.unwrap(),	1✔
563	vec![TimeInterval::default(); 4],	1✔
564	Default::default(),	1✔
565	)	1✔
566	.unwrap()],	1✔
567	SpatialReference::from_str("EPSG:3857").unwrap(),	1✔
568	)	1✔
569	.boxed();	1✔
570		1✔
571	let mut exe_ctc = MockExecutionContext::test_default();	1✔
572	let ndvi_id = add_ndvi_dataset(&mut exe_ctc);	1✔
573
574	let operator = RasterVectorJoin {	1✔
575	params: RasterVectorJoinParams {	1✔
576	names: vec!["ndvi".to_string()],	1✔
577	feature_aggregation: FeatureAggregationMethod::First,	1✔
578	temporal_aggregation: TemporalAggregationMethod::Mean,	1✔
579	},	1✔
580	sources: SingleVectorMultipleRasterSources {	1✔
581	vector: point_source,	1✔
582	rasters: vec![ndvi_source(ndvi_id.clone())],	1✔
583	},	1✔
584	}	1✔
585	.boxed()	1✔
586	.initialize(&exe_ctc)	1✔
587	.await;	×
588
589	assert!(matches!(	1✔
590	operator,	1✔
591	Err(Error::InvalidSpatialReference {
592	expected: _,
593	found: _,
594	})
595	));
596	}
597	}

geo-engine / geoengine / 3929938005

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

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