5006008836

Build Type

push

github

Committed by GitHub

Commit Message

Merge #785 #787

Run Details

936 of 936 new or added lines in 50 files covered. (100.0%)

96010 of 107707 relevant lines covered (89.14%)

72676.46 hits per line

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

86.62

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

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

use crate::engine::{
    CanonicOperatorName, ExecutionContext, InitializedRasterOperator, InitializedVectorOperator,
    Operator, OperatorName, SingleVectorMultipleRasterSources, TypedVectorQueryProcessor,
    VectorColumnInfo, VectorOperator, VectorQueryProcessor, VectorResultDescriptor,
    WorkflowOperatorPath,
};
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 geoengine_datatypes::collections::VectorDataType;
use geoengine_datatypes::primitives::FeatureDataType;
use geoengine_datatypes::raster::{Pixel, RasterDataType};
use serde::{Deserialize, Serialize};
use snafu::ensure;

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>,
        path: WorkflowOperatorPath,
        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 name = CanonicOperatorName::from(&self);

        let vector_source = self
            .sources
            .vector
            .initialize(path.clone_and_append(0), 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 = futures::future::try_join_all(
            self.sources
                .rasters
                .into_iter()
                .enumerate()
                .map(|(i, op)| op.initialize(path.clone_and_append(i as u8 + 1), context)),
        )
        .await?;

        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 {
            name,
            result_descriptor,
            vector_source,
            raster_sources,
            state: params,
        }
        .boxed())
    }

    span_fn!(RasterVectorJoin);
}

pub struct InitializedRasterVectorJoin {
    name: CanonicOperatorName,
    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),
        })
    }

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

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

geo-engine / geoengine / 5006008836

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