19241805651

Committed 10 Nov 2025 06:22PM UTC coverage: 88.832%. First build

Build # 19241805651

Build Type

Pull #1083

github

Committed by

web-flow

Commit Message

Merge dac631b93 into 113de40ca

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

Run Details

6213 of 7052 new or added lines in 71 files covered. (88.1%)

116189 of 130797 relevant lines covered (88.83%)

496404.71 hits per line

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

95.61

/operators/src/plot/temporal_vector_line_plot.rs

use crate::engine::{
    CanonicOperatorName, ExecutionContext, InitializedPlotOperator, InitializedSources,
    InitializedVectorOperator, Operator, OperatorName, PlotOperator, PlotQueryProcessor,
    PlotResultDescriptor, QueryContext, QueryProcessor, SingleVectorSource,
    TypedPlotQueryProcessor, VectorColumnInfo, VectorQueryProcessor, WorkflowOperatorPath,
};
use crate::error;
use crate::optimization::OptimizationError;
use crate::util::Result;
use async_trait::async_trait;
use futures::TryStreamExt;
use geoengine_datatypes::primitives::SpatialResolution;
use geoengine_datatypes::{
    collections::{FeatureCollection, FeatureCollectionInfos},
    plots::{DataPoint, MultiLineChart, Plot, PlotData},
    primitives::{
        ColumnSelection, FeatureDataType, Geometry, Measurement, PlotQueryRectangle, TimeInterval,
    },
    util::arrow::ArrowTyped,
};
use serde::{Deserialize, Serialize};
use snafu::ensure;
use std::collections::HashMap;
use std::{
    cmp::Ordering,
    collections::hash_map::Entry::{Occupied, Vacant},
};

pub const FEATURE_ATTRIBUTE_OVER_TIME_NAME: &str = "Feature Attribute over Time";
const MAX_FEATURES: usize = 20;

/// A plot that shows the value of an feature attribute over time.
pub type FeatureAttributeValuesOverTime =
    Operator<FeatureAttributeValuesOverTimeParams, SingleVectorSource>;

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

/// The parameter spec for `FeatureAttributeValuesOverTime`
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct FeatureAttributeValuesOverTimeParams {
    pub id_column: String,
    pub value_column: String,
}

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

        let initialized_source = self
            .sources
            .initialize_sources(path.clone(), context)
            .await?;
        let source = initialized_source.vector;
        let result_descriptor = source.result_descriptor();
        let columns: &HashMap<String, VectorColumnInfo> = &result_descriptor.columns;

        ensure!(
            columns.contains_key(&self.params.id_column),
            error::ColumnDoesNotExist {
                column: self.params.id_column.clone()
            }
        );

        ensure!(
            columns.contains_key(&self.params.value_column),
            error::ColumnDoesNotExist {
                column: self.params.value_column.clone()
            }
        );

        let id_type = columns
            .get(&self.params.id_column)
            .expect("checked")
            .data_type;
        let value_type = columns
            .get(&self.params.value_column)
            .expect("checked")
            .data_type;

        // TODO: ensure column is really an id
        ensure!(
            id_type == FeatureDataType::Text
                || id_type == FeatureDataType::Int
                || id_type == FeatureDataType::Category,
            error::InvalidFeatureDataType,
        );

        ensure!(
            value_type.is_numeric() || value_type == FeatureDataType::Category,
            error::InvalidFeatureDataType,
        );

        let in_desc = source.result_descriptor().clone();

        Ok(InitializedFeatureAttributeValuesOverTime {
            name,
            result_descriptor: in_desc.into(),
            vector_source: source,
            state: self.params,
        }
        .boxed())
    }

    span_fn!(FeatureAttributeValuesOverTime);
}

/// The initialization of `FeatureAttributeValuesOverTime`
pub struct InitializedFeatureAttributeValuesOverTime {
    name: CanonicOperatorName,
    result_descriptor: PlotResultDescriptor,
    vector_source: Box<dyn InitializedVectorOperator>,
    state: FeatureAttributeValuesOverTimeParams,
}

impl InitializedPlotOperator for InitializedFeatureAttributeValuesOverTime {
    fn query_processor(&self) -> Result<TypedPlotQueryProcessor> {
        let input_processor = self.vector_source.query_processor()?;

        let processor = call_on_generic_vector_processor!(input_processor, features => {
            FeatureAttributeValuesOverTimeQueryProcessor { params: self.state.clone(), features }.boxed()
        });

        Ok(TypedPlotQueryProcessor::JsonVega(processor))
    }

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

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

    fn optimize(
        &self,
        target_resolution: SpatialResolution,
    ) -> Result<Box<dyn PlotOperator>, OptimizationError> {
        Ok(FeatureAttributeValuesOverTime {
            params: self.state.clone(),
            sources: SingleVectorSource {
                vector: self.vector_source.optimize(target_resolution)?,
            },
        }
        .boxed())
    }
}

/// A query processor that calculates the `TemporalVectorLinePlot` on its input.
pub struct FeatureAttributeValuesOverTimeQueryProcessor<G>
where
    G: Geometry + ArrowTyped + Sync + Send + 'static,
{
    params: FeatureAttributeValuesOverTimeParams,
    features: Box<dyn VectorQueryProcessor<VectorType = FeatureCollection<G>>>,
}

#[async_trait]
impl<G> PlotQueryProcessor for FeatureAttributeValuesOverTimeQueryProcessor<G>
where
    G: Geometry + ArrowTyped + Sync + Send + 'static,
{
    type OutputFormat = PlotData;

    fn plot_type(&self) -> &'static str {
        FEATURE_ATTRIBUTE_OVER_TIME_NAME
    }

    async fn plot_query<'a>(
        &'a self,
        query: PlotQueryRectangle,
        ctx: &'a dyn QueryContext,
    ) -> Result<Self::OutputFormat> {
        let values = FeatureAttributeValues::<MAX_FEATURES>::default();

        let query = query.select_attributes(ColumnSelection::all());

        let values = self
            .features
            .query(query, ctx)
            .await?
            .try_fold(values, |mut acc, features| async move {
                let ids = features.data(&self.params.id_column)?;
                let values = features.data(&self.params.value_column)?;

                for ((id, value), &time) in ids
                    .strings_iter()
                    .zip(values.float_options_iter())
                    .zip(features.time_intervals())
                {
                    if id.is_empty() || value.is_none() {
                        continue;
                    }

                    let value = value.expect("checked above");

                    acc.add(id, (time, value));
                }

                Ok(acc)
            })
            .await?;

        let data_points = values.get_data_points();
        let measurement = Measurement::Unitless; // TODO: attach actual unit if we know it
        MultiLineChart::new(data_points, measurement)
            .to_vega_embeddable(false)
            .map_err(Into::into)
    }
}

struct TemporalValue {
    pub time: TimeInterval,
    pub value: f64,
}

impl From<(TimeInterval, f64)> for TemporalValue {
    fn from(value: (TimeInterval, f64)) -> Self {
        Self {
            time: value.0,
            value: value.1,
        }
    }
}

struct FeatureAttributeValues<const LENGTH: usize> {
    values: HashMap<String, Vec<TemporalValue>>,
}

impl<const LENGTH: usize> Default for FeatureAttributeValues<LENGTH> {
    fn default() -> Self {
        Self {
            values: HashMap::with_capacity(LENGTH),
        }
    }
}

impl<const LENGTH: usize> FeatureAttributeValues<LENGTH> {
    /// Add value to the data structure. If `id` is new and there are already `LENGTH` existing
    /// `id`-entries, the value is ignored
    pub fn add<V>(&mut self, id: String, value: V)
    where
        V: Into<TemporalValue>,
    {
        let len = self.values.len();

        match self.values.entry(id) {
            Occupied(mut entry) => entry.get_mut().push(value.into()),
            Vacant(entry) => {
                if len < LENGTH {
                    entry.insert(vec![value.into()]);
                }
            }
        }
    }

    pub fn get_data_points(mut self) -> Vec<DataPoint> {
        let mut data = self
            .values
            .drain()
            .flat_map(|(id, values)| {
                values.into_iter().map(move |value| DataPoint {
                    series: id.clone(),
                    time: value.time.start(),
                    value: value.value,
                })
            })
            .collect::<Vec<_>>();

        data.sort_unstable_by(|a, b| match a.series.cmp(&b.series) {
            Ordering::Equal => a.time.cmp(&b.time),
            other => other,
        });
        data
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        engine::{ChunkByteSize, MockExecutionContext, VectorOperator},
        mock::MockFeatureCollectionSource,
    };
    use geoengine_datatypes::primitives::PlotQueryRectangle;
    use geoengine_datatypes::primitives::{CacheHint, PlotSeriesSelection};
    use geoengine_datatypes::util::test::TestDefault;
    use geoengine_datatypes::{
        collections::MultiPointCollection,
        plots::PlotMetaData,
        primitives::{BoundingBox2D, DateTime, FeatureData, MultiPoint, TimeInterval},
    };
    use serde_json::{Value, json};

    #[tokio::test]
    #[allow(clippy::too_many_lines)]
    async fn plot() {
        let point_source = MockFeatureCollectionSource::single(
            MultiPointCollection::from_data(
                MultiPoint::many(vec![
                    vec![(-13.95, 20.05)],
                    vec![(-14.05, 20.05)],
                    vec![(-13.95, 20.05)],
                ])
                .unwrap(),
                vec![
                    TimeInterval::new_unchecked(
                        DateTime::new_utc(2014, 1, 1, 0, 0, 0),
                        DateTime::new_utc(2014, 2, 1, 0, 0, 0),
                    ),
                    TimeInterval::new_unchecked(
                        DateTime::new_utc(2014, 1, 1, 0, 0, 0),
                        DateTime::new_utc(2014, 3, 1, 0, 0, 0),
                    ),
                    TimeInterval::new_unchecked(
                        DateTime::new_utc(2014, 2, 1, 0, 0, 0),
                        DateTime::new_utc(2014, 3, 1, 0, 0, 0),
                    ),
                ],
                [
                    (
                        "id".to_string(),
                        FeatureData::Text(vec!["S0".to_owned(), "S1".to_owned(), "S0".to_owned()]),
                    ),
                    ("value".to_string(), FeatureData::Float(vec![0., 2., 1.])),
                ]
                .iter()
                .cloned()
                .collect(),
                CacheHint::default(),
            )
            .unwrap(),
        )
        .boxed();

        let exe_ctc = MockExecutionContext::test_default();

        let operator = FeatureAttributeValuesOverTime {
            params: FeatureAttributeValuesOverTimeParams {
                id_column: "id".to_owned(),
                value_column: "value".to_owned(),
            },
            sources: point_source.into(),
        };

        let operator = operator
            .boxed()
            .initialize(WorkflowOperatorPath::initialize_root(), &exe_ctc)
            .await
            .unwrap();

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

        let result = query_processor
            .plot_query(
                PlotQueryRectangle::new(
                    BoundingBox2D::new((-180., -90.).into(), (180., 90.).into()).unwrap(),
                    TimeInterval::default(),
                    PlotSeriesSelection::all(),
                ),
                &exe_ctc.mock_query_context(ChunkByteSize::MIN),
            )
            .await
            .unwrap();

        assert!(matches!(result.metadata, PlotMetaData::None));

        let vega_json: Value = serde_json::from_str(&result.vega_string).unwrap();

        assert_eq!(
            vega_json,
            json!({
                "$schema": "https://vega.github.io/schema/vega-lite/v4.17.0.json",
                "data": {
                    "values": [{
                        "x": "2014-01-01T00:00:00+00:00",
                        "y": 0.0,
                        "series": "S0"
                    }, {
                        "x": "2014-02-01T00:00:00+00:00",
                        "y": 1.0,
                        "series": "S0"
                    }, {
                        "x": "2014-01-01T00:00:00+00:00",
                        "y": 2.0,
                        "series": "S1"
                    }]
                },
                "description": "Multi Line Chart",
                "encoding": {
                    "x": {
                        "field": "x",
                        "title": "Time",
                        "type": "temporal"
                    },
                    "y": {
                        "field": "y",
                        "title": "",
                        "type": "quantitative"
                    },
                    "color": {
                        "field": "series",
                        "scale": {
                            "scheme": "category20"
                        }
                    }
                },
                "mark": {
                    "type": "line",
                    "line": true,
                    "point": true
                }
            })
        );
    }

    #[tokio::test]
    #[allow(clippy::too_many_lines)]
    async fn plot_with_nulls() {
        let point_source = MockFeatureCollectionSource::single(
            MultiPointCollection::from_data(
                MultiPoint::many(vec![
                    vec![(-13.95, 20.05)],
                    vec![(-14.05, 20.05)],
                    vec![(-13.95, 20.05)],
                    vec![(-14.05, 20.05)],
                    vec![(-13.95, 20.05)],
                ])
                .unwrap(),
                vec![
                    TimeInterval::new_unchecked(
                        DateTime::new_utc(2014, 1, 1, 0, 0, 0),
                        DateTime::new_utc(2014, 2, 1, 0, 0, 0),
                    ),
                    TimeInterval::new_unchecked(
                        DateTime::new_utc(2014, 1, 1, 0, 0, 0),
                        DateTime::new_utc(2014, 3, 1, 0, 0, 0),
                    ),
                    TimeInterval::new_unchecked(
                        DateTime::new_utc(2014, 2, 1, 0, 0, 0),
                        DateTime::new_utc(2014, 3, 1, 0, 0, 0),
                    ),
                    TimeInterval::new_unchecked(
                        DateTime::new_utc(2014, 1, 1, 0, 0, 0),
                        DateTime::new_utc(2014, 3, 1, 0, 0, 0),
                    ),
                    TimeInterval::new_unchecked(
                        DateTime::new_utc(2014, 2, 1, 0, 0, 0),
                        DateTime::new_utc(2014, 3, 1, 0, 0, 0),
                    ),
                ],
                [
                    (
                        "id".to_string(),
                        FeatureData::NullableText(vec![
                            Some("S0".to_owned()),
                            Some("S1".to_owned()),
                            Some("S0".to_owned()),
                            None,
                            Some("S2".to_owned()),
                        ]),
                    ),
                    (
                        "value".to_string(),
                        FeatureData::NullableFloat(vec![
                            Some(0.),
                            Some(2.),
                            Some(1.),
                            Some(3.),
                            None,
                        ]),
                    ),
                ]
                .iter()
                .cloned()
                .collect(),
                CacheHint::default(),
            )
            .unwrap(),
        )
        .boxed();

        let exe_ctc = MockExecutionContext::test_default();

        let operator = FeatureAttributeValuesOverTime {
            params: FeatureAttributeValuesOverTimeParams {
                id_column: "id".to_owned(),
                value_column: "value".to_owned(),
            },
            sources: point_source.into(),
        };

        let operator = operator
            .boxed()
            .initialize(WorkflowOperatorPath::initialize_root(), &exe_ctc)
            .await
            .unwrap();

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

        let result = query_processor
            .plot_query(
                PlotQueryRectangle::new(
                    BoundingBox2D::new((-180., -90.).into(), (180., 90.).into()).unwrap(),
                    TimeInterval::default(),
                    PlotSeriesSelection::all(),
                ),
                &exe_ctc.mock_query_context(ChunkByteSize::MIN),
            )
            .await
            .unwrap();

        assert!(matches!(result.metadata, PlotMetaData::None));

        let vega_json: Value = serde_json::from_str(&result.vega_string).unwrap();

        assert_eq!(
            vega_json,
            json!({
                "$schema": "https://vega.github.io/schema/vega-lite/v4.17.0.json",
                "data": {
                    "values": [{
                        "x": "2014-01-01T00:00:00+00:00",
                        "y": 0.0,
                        "series": "S0"
                    }, {
                        "x": "2014-02-01T00:00:00+00:00",
                        "y": 1.0,
                        "series": "S0"
                    }, {
                        "x": "2014-01-01T00:00:00+00:00",
                        "y": 2.0,
                        "series": "S1"
                    }]
                },
                "description": "Multi Line Chart",
                "encoding": {
                    "x": {
                        "field": "x",
                        "title": "Time",
                        "type": "temporal"
                    },
                    "y": {
                        "field": "y",
                        "title": "",
                        "type": "quantitative"
                    },
                    "color": {
                        "field": "series",
                        "scale": {
                            "scheme": "category20"
                        }
                    }
                },
                "mark": {
                    "type": "line",
                    "line": true,
                    "point": true
                }
            })
        );
    }

    #[tokio::test]
    #[allow(clippy::too_many_lines)]
    async fn plot_with_duplicates() {
        let point_source = MockFeatureCollectionSource::single(
            MultiPointCollection::from_data(
                MultiPoint::many(vec![
                    vec![(-13.95, 20.05)],
                    vec![(-14.05, 20.05)],
                    vec![(-13.95, 20.05)],
                    vec![(-13.95, 20.05)],
                ])
                .unwrap(),
                vec![
                    TimeInterval::new_unchecked(
                        DateTime::new_utc(2014, 1, 1, 0, 0, 0),
                        DateTime::new_utc(2014, 2, 1, 0, 0, 0),
                    ),
                    TimeInterval::new_unchecked(
                        DateTime::new_utc(2014, 1, 1, 0, 0, 0),
                        DateTime::new_utc(2014, 3, 1, 0, 0, 0),
                    ),
                    TimeInterval::new_unchecked(
                        DateTime::new_utc(2014, 2, 1, 0, 0, 0),
                        DateTime::new_utc(2014, 3, 1, 0, 0, 0),
                    ),
                    TimeInterval::new_unchecked(
                        DateTime::new_utc(2014, 2, 1, 0, 0, 0),
                        DateTime::new_utc(2014, 3, 1, 0, 0, 0),
                    ),
                ],
                [
                    (
                        "id".to_string(),
                        FeatureData::Text(vec![
                            "S0".to_owned(),
                            "S1".to_owned(),
                            "S0".to_owned(),
                            "S0".to_owned(),
                        ]),
                    ),
                    (
                        "value".to_string(),
                        FeatureData::Float(vec![0., 2., 1., 1.]),
                    ),
                ]
                .iter()
                .cloned()
                .collect(),
                CacheHint::default(),
            )
            .unwrap(),
        )
        .boxed();

        let exe_ctx = MockExecutionContext::test_default();

        let operator = FeatureAttributeValuesOverTime {
            params: FeatureAttributeValuesOverTimeParams {
                id_column: "id".to_owned(),
                value_column: "value".to_owned(),
            },
            sources: point_source.into(),
        };

        let operator = operator
            .boxed()
            .initialize(WorkflowOperatorPath::initialize_root(), &exe_ctx)
            .await
            .unwrap();

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

        let result = query_processor
            .plot_query(
                PlotQueryRectangle::new(
                    BoundingBox2D::new((-180., -90.).into(), (180., 90.).into()).unwrap(),
                    TimeInterval::default(),
                    PlotSeriesSelection::all(),
                ),
                &exe_ctx.mock_query_context(ChunkByteSize::MIN),
            )
            .await
            .unwrap();

        assert!(matches!(result.metadata, PlotMetaData::None));

        let vega_json: Value = serde_json::from_str(&result.vega_string).unwrap();

        assert_eq!(
            vega_json,
            json!({
                "$schema": "https://vega.github.io/schema/vega-lite/v4.17.0.json",
                "data": {
                    "values": [{
                        "x": "2014-01-01T00:00:00+00:00",
                        "y": 0.0,
                        "series": "S0"
                    }, {
                        "x": "2014-02-01T00:00:00+00:00",
                        "y": 1.0,
                        "series": "S0"
                    }, {
                        "x": "2014-02-01T00:00:00+00:00",
                        "y": 1.0,
                        "series": "S0"
                    }, {
                        "x": "2014-01-01T00:00:00+00:00",
                        "y": 2.0,
                        "series": "S1"
                    }]
                },
                "description": "Multi Line Chart",
                "encoding": {
                    "x": {
                        "field": "x",
                        "title": "Time",
                        "type": "temporal"
                    },
                    "y": {
                        "field": "y",
                        "title": "",
                        "type": "quantitative"
                    },
                    "color": {
                        "field": "series",
                        "scale": {
                            "scheme": "category20"
                        }
                    }
                },
                "mark": {
                    "type": "line",
                    "line": true,
                    "point": true
                }
            })
        );
    }
}

1	use crate::engine::{
2	CanonicOperatorName, ExecutionContext, InitializedPlotOperator, InitializedSources,
3	InitializedVectorOperator, Operator, OperatorName, PlotOperator, PlotQueryProcessor,
4	PlotResultDescriptor, QueryContext, QueryProcessor, SingleVectorSource,
5	TypedPlotQueryProcessor, VectorColumnInfo, VectorQueryProcessor, WorkflowOperatorPath,
6	};
7	use crate::error;
8	use crate::optimization::OptimizationError;
9	use crate::util::Result;
10	use async_trait::async_trait;
11	use futures::TryStreamExt;
12	use geoengine_datatypes::primitives::SpatialResolution;
13	use geoengine_datatypes::{
14	collections::{FeatureCollection, FeatureCollectionInfos},
15	plots::{DataPoint, MultiLineChart, Plot, PlotData},
16	primitives::{
17	ColumnSelection, FeatureDataType, Geometry, Measurement, PlotQueryRectangle, TimeInterval,
18	},
19	util::arrow::ArrowTyped,
20	};
21	use serde::{Deserialize, Serialize};
22	use snafu::ensure;
23	use std::collections::HashMap;
24	use std::{
25	cmp::Ordering,
26	collections::hash_map::Entry::{Occupied, Vacant},
27	};
28
29	pub const FEATURE_ATTRIBUTE_OVER_TIME_NAME: &str = "Feature Attribute over Time";
30	const MAX_FEATURES: usize = 20;
31
32	/// A plot that shows the value of an feature attribute over time.
33	pub type FeatureAttributeValuesOverTime =
34	Operator<FeatureAttributeValuesOverTimeParams, SingleVectorSource>;
35
36	impl OperatorName for FeatureAttributeValuesOverTime {
37	const TYPE_NAME: &'static str = "FeatureAttributeValuesOverTime";
38	}
39
40	/// The parameter spec for `FeatureAttributeValuesOverTime`
41	#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
42	#[serde(rename_all = "camelCase")]
43	pub struct FeatureAttributeValuesOverTimeParams {
44	pub id_column: String,
45	pub value_column: String,
46	}
47
48	#[typetag::serde]	×
49	#[async_trait]
50	impl PlotOperator for FeatureAttributeValuesOverTime {
51	async fn _initialize(
52	self: Box<Self>,
53	path: WorkflowOperatorPath,
54	context: &dyn ExecutionContext,
55	) -> Result<Box<dyn InitializedPlotOperator>> {	3✔
56	let name = CanonicOperatorName::from(&self);
57
58	let initialized_source = self
59	.sources
60	.initialize_sources(path.clone(), context)
61	.await?;
62	let source = initialized_source.vector;
63	let result_descriptor = source.result_descriptor();
64	let columns: &HashMap<String, VectorColumnInfo> = &result_descriptor.columns;
65
66	ensure!(
67	columns.contains_key(&self.params.id_column),
68	error::ColumnDoesNotExist {
69	column: self.params.id_column.clone()
70	}
71	);
72
73	ensure!(
74	columns.contains_key(&self.params.value_column),
75	error::ColumnDoesNotExist {
76	column: self.params.value_column.clone()
77	}
78	);
79
80	let id_type = columns
81	.get(&self.params.id_column)
82	.expect("checked")
83	.data_type;
84	let value_type = columns
85	.get(&self.params.value_column)
86	.expect("checked")
87	.data_type;
88
89	// TODO: ensure column is really an id
90	ensure!(
91	id_type == FeatureDataType::Text
92	\|\| id_type == FeatureDataType::Int
93	\|\| id_type == FeatureDataType::Category,
94	error::InvalidFeatureDataType,
95	);
96
97	ensure!(
98	value_type.is_numeric() \|\| value_type == FeatureDataType::Category,
99	error::InvalidFeatureDataType,
100	);
101
102	let in_desc = source.result_descriptor().clone();
103
104	Ok(InitializedFeatureAttributeValuesOverTime {
105	name,
106	result_descriptor: in_desc.into(),
107	vector_source: source,
108	state: self.params,
109	}
110	.boxed())
111	}	3✔
112
113	span_fn!(FeatureAttributeValuesOverTime);
114	}
115
116	/// The initialization of `FeatureAttributeValuesOverTime`
117	pub struct InitializedFeatureAttributeValuesOverTime {
118	name: CanonicOperatorName,
119	result_descriptor: PlotResultDescriptor,
120	vector_source: Box<dyn InitializedVectorOperator>,
121	state: FeatureAttributeValuesOverTimeParams,
122	}
123
124	impl InitializedPlotOperator for InitializedFeatureAttributeValuesOverTime {
125	fn query_processor(&self) -> Result<TypedPlotQueryProcessor> {	3✔
126	let input_processor = self.vector_source.query_processor()?;	3✔
127
128	let processor = call_on_generic_vector_processor!(input_processor, features => {	3✔
129	FeatureAttributeValuesOverTimeQueryProcessor { params: self.state.clone(), features }.boxed()	×
130	});
131
132	Ok(TypedPlotQueryProcessor::JsonVega(processor))	3✔
133	}	3✔
134
135	fn result_descriptor(&self) -> &PlotResultDescriptor {	×
136	&self.result_descriptor	×
137	}	×
138
139	fn canonic_name(&self) -> CanonicOperatorName {	×
140	self.name.clone()	×
141	}	×
142
NEW 143	fn optimize(	×
NEW 144	&self,	×
NEW 145	target_resolution: SpatialResolution,	×
NEW 146	) -> Result<Box<dyn PlotOperator>, OptimizationError> {	×
147	Ok(FeatureAttributeValuesOverTime {
NEW 148	params: self.state.clone(),	×
149	sources: SingleVectorSource {
NEW 150	vector: self.vector_source.optimize(target_resolution)?,	×
151	},
152	}
NEW 153	.boxed())	×
NEW 154	}	×
155	}
156
157	/// A query processor that calculates the `TemporalVectorLinePlot` on its input.
158	pub struct FeatureAttributeValuesOverTimeQueryProcessor<G>
159	where
160	G: Geometry + ArrowTyped + Sync + Send + 'static,
161	{
162	params: FeatureAttributeValuesOverTimeParams,
163	features: Box<dyn VectorQueryProcessor<VectorType = FeatureCollection<G>>>,
164	}
165
166	#[async_trait]
167	impl<G> PlotQueryProcessor for FeatureAttributeValuesOverTimeQueryProcessor<G>
168	where
169	G: Geometry + ArrowTyped + Sync + Send + 'static,
170	{
171	type OutputFormat = PlotData;
172
173	fn plot_type(&self) -> &'static str {	×
174	FEATURE_ATTRIBUTE_OVER_TIME_NAME	×
175	}	×
176
177	async fn plot_query<'a>(
178	&'a self,
179	query: PlotQueryRectangle,
180	ctx: &'a dyn QueryContext,
181	) -> Result<Self::OutputFormat> {	3✔
182	let values = FeatureAttributeValues::<MAX_FEATURES>::default();
183
184	let query = query.select_attributes(ColumnSelection::all());
185
186	let values = self
187	.features
188	.query(query, ctx)
189	.await?
190	.try_fold(values, \|mut acc, features\| async move {	3✔
191	let ids = features.data(&self.params.id_column)?;	3✔
192	let values = features.data(&self.params.value_column)?;	3✔
193
194	for ((id, value), &time) in ids	12✔
195	.strings_iter()	3✔
196	.zip(values.float_options_iter())	3✔
197	.zip(features.time_intervals())	3✔
198	{
199	if id.is_empty() \|\| value.is_none() {	12✔
200	continue;	2✔
201	}	10✔
202
203	let value = value.expect("checked above");	10✔
204
205	acc.add(id, (time, value));	10✔
206	}
207
208	Ok(acc)	3✔
209	})	6✔
210	.await?;
211
212	let data_points = values.get_data_points();
213	let measurement = Measurement::Unitless; // TODO: attach actual unit if we know it
214	MultiLineChart::new(data_points, measurement)
215	.to_vega_embeddable(false)
216	.map_err(Into::into)
217	}	3✔
218	}
219
220	struct TemporalValue {
221	pub time: TimeInterval,
222	pub value: f64,
223	}
224
225	impl From<(TimeInterval, f64)> for TemporalValue {
226	fn from(value: (TimeInterval, f64)) -> Self {	10✔
227	Self {	10✔
228	time: value.0,	10✔
229	value: value.1,	10✔
230	}	10✔
231	}	10✔
232	}
233
234	struct FeatureAttributeValues<const LENGTH: usize> {
235	values: HashMap<String, Vec<TemporalValue>>,
236	}
237
238	impl<const LENGTH: usize> Default for FeatureAttributeValues<LENGTH> {
239	fn default() -> Self {	3✔
240	Self {	3✔
241	values: HashMap::with_capacity(LENGTH),	3✔
242	}	3✔
243	}	3✔
244	}
245
246	impl<const LENGTH: usize> FeatureAttributeValues<LENGTH> {
247	/// Add value to the data structure. If `id` is new and there are already `LENGTH` existing
248	/// `id`-entries, the value is ignored
249	pub fn add<V>(&mut self, id: String, value: V)	10✔
250	where	10✔
251	V: Into<TemporalValue>,	10✔
252	{
253	let len = self.values.len();	10✔
254
255	match self.values.entry(id) {	10✔
256	Occupied(mut entry) => entry.get_mut().push(value.into()),	4✔
257	Vacant(entry) => {	6✔
258	if len < LENGTH {	6✔
259	entry.insert(vec![value.into()]);	6✔
260	}	6✔
261	}
262	}
263	}	10✔
264
265	pub fn get_data_points(mut self) -> Vec<DataPoint> {	3✔
266	let mut data = self	3✔
267	.values	3✔
268	.drain()	3✔
269	.flat_map(\|(id, values)\| {	6✔
270	values.into_iter().map(move \|value\| DataPoint {	6✔
271	series: id.clone(),	10✔
272	time: value.time.start(),	10✔
273	value: value.value,	10✔
274	})	10✔
275	})	6✔
276	.collect::<Vec<_>>();	3✔
277
278	data.sort_unstable_by(\|a, b\| match a.series.cmp(&b.series) {	9✔
279	Ordering::Equal => a.time.cmp(&b.time),	4✔
280	other => other,	5✔
281	});	9✔
282	data	3✔
283	}	3✔
284	}
285
286	#[cfg(test)]
287	mod tests {
288	use super::*;
289	use crate::{
290	engine::{ChunkByteSize, MockExecutionContext, VectorOperator},
291	mock::MockFeatureCollectionSource,
292	};
293	use geoengine_datatypes::primitives::PlotQueryRectangle;
294	use geoengine_datatypes::primitives::{CacheHint, PlotSeriesSelection};
295	use geoengine_datatypes::util::test::TestDefault;
296	use geoengine_datatypes::{
297	collections::MultiPointCollection,
298	plots::PlotMetaData,
299	primitives::{BoundingBox2D, DateTime, FeatureData, MultiPoint, TimeInterval},
300	};
301	use serde_json::{Value, json};
302
303	#[tokio::test]
304	#[allow(clippy::too_many_lines)]
305	async fn plot() {	1✔
306	let point_source = MockFeatureCollectionSource::single(	1✔
307	MultiPointCollection::from_data(	1✔
308	MultiPoint::many(vec![	1✔
309	vec![(-13.95, 20.05)],	1✔
310	vec![(-14.05, 20.05)],	1✔
311	vec![(-13.95, 20.05)],	1✔
312	])
313	.unwrap(),	1✔
314	vec![	1✔
315	TimeInterval::new_unchecked(	1✔
316	DateTime::new_utc(2014, 1, 1, 0, 0, 0),	1✔
317	DateTime::new_utc(2014, 2, 1, 0, 0, 0),	1✔
318	),
319	TimeInterval::new_unchecked(	1✔
320	DateTime::new_utc(2014, 1, 1, 0, 0, 0),	1✔
321	DateTime::new_utc(2014, 3, 1, 0, 0, 0),	1✔
322	),
323	TimeInterval::new_unchecked(	1✔
324	DateTime::new_utc(2014, 2, 1, 0, 0, 0),	1✔
325	DateTime::new_utc(2014, 3, 1, 0, 0, 0),	1✔
326	),
327	],
328	[	1✔
329	(	1✔
330	"id".to_string(),	1✔
331	FeatureData::Text(vec!["S0".to_owned(), "S1".to_owned(), "S0".to_owned()]),	1✔
332	),	1✔
333	("value".to_string(), FeatureData::Float(vec![0., 2., 1.])),	1✔
334	]	1✔
335	.iter()	1✔
336	.cloned()	1✔
337	.collect(),	1✔
338	CacheHint::default(),	1✔
339	)
340	.unwrap(),	1✔
341	)
342	.boxed();	1✔
343
344	let exe_ctc = MockExecutionContext::test_default();	1✔
345
346	let operator = FeatureAttributeValuesOverTime {	1✔
347	params: FeatureAttributeValuesOverTimeParams {	1✔
348	id_column: "id".to_owned(),	1✔
349	value_column: "value".to_owned(),	1✔
350	},	1✔
351	sources: point_source.into(),	1✔
352	};	1✔
353
354	let operator = operator	1✔
355	.boxed()	1✔
356	.initialize(WorkflowOperatorPath::initialize_root(), &exe_ctc)	1✔
357	.await	1✔
358	.unwrap();	1✔
359
360	let query_processor = operator.query_processor().unwrap().json_vega().unwrap();	1✔
361
362	let result = query_processor	1✔
363	.plot_query(	1✔
364	PlotQueryRectangle::new(	1✔
365	BoundingBox2D::new((-180., -90.).into(), (180., 90.).into()).unwrap(),	1✔
366	TimeInterval::default(),	1✔
367	PlotSeriesSelection::all(),	1✔
368	),	1✔
369	&exe_ctc.mock_query_context(ChunkByteSize::MIN),	1✔
370	)	1✔
371	.await	1✔
372	.unwrap();	1✔
373
374	assert!(matches!(result.metadata, PlotMetaData::None));	1✔
375
376	let vega_json: Value = serde_json::from_str(&result.vega_string).unwrap();	1✔
377
378	assert_eq!(	1✔
379	vega_json,	1✔
380	json!({	1✔
381	"$schema": "https://vega.github.io/schema/vega-lite/v4.17.0.json",	1✔
382	"data": {	1✔
383	"values": [{	1✔
384	"x": "2014-01-01T00:00:00+00:00",	1✔
385	"y": 0.0,	1✔
386	"series": "S0"	1✔
387	}, {	1✔
388	"x": "2014-02-01T00:00:00+00:00",	1✔
389	"y": 1.0,	1✔
390	"series": "S0"	1✔
391	}, {	1✔
392	"x": "2014-01-01T00:00:00+00:00",	1✔
393	"y": 2.0,	1✔
394	"series": "S1"	1✔
395	}]	1✔
396	},	1✔
397	"description": "Multi Line Chart",	1✔
398	"encoding": {	1✔
399	"x": {	1✔
400	"field": "x",	1✔
401	"title": "Time",	1✔
402	"type": "temporal"	1✔
403	},	1✔
404	"y": {	1✔
405	"field": "y",	1✔
406	"title": "",	1✔
407	"type": "quantitative"	1✔
408	},	1✔
409	"color": {	1✔
410	"field": "series",	1✔
411	"scale": {	1✔
412	"scheme": "category20"	1✔
413	}	1✔
414	}	1✔
415	},	1✔
416	"mark": {	1✔
417	"type": "line",	1✔
418	"line": true,	1✔
419	"point": true	1✔
420	}	1✔
421	})	1✔
422	);	1✔
423	}	1✔
424
425	#[tokio::test]
426	#[allow(clippy::too_many_lines)]
427	async fn plot_with_nulls() {	1✔
428	let point_source = MockFeatureCollectionSource::single(	1✔
429	MultiPointCollection::from_data(	1✔
430	MultiPoint::many(vec![	1✔
431	vec![(-13.95, 20.05)],	1✔
432	vec![(-14.05, 20.05)],	1✔
433	vec![(-13.95, 20.05)],	1✔
434	vec![(-14.05, 20.05)],	1✔
435	vec![(-13.95, 20.05)],	1✔
436	])
437	.unwrap(),	1✔
438	vec![	1✔
439	TimeInterval::new_unchecked(	1✔
440	DateTime::new_utc(2014, 1, 1, 0, 0, 0),	1✔
441	DateTime::new_utc(2014, 2, 1, 0, 0, 0),	1✔
442	),
443	TimeInterval::new_unchecked(	1✔
444	DateTime::new_utc(2014, 1, 1, 0, 0, 0),	1✔
445	DateTime::new_utc(2014, 3, 1, 0, 0, 0),	1✔
446	),
447	TimeInterval::new_unchecked(	1✔
448	DateTime::new_utc(2014, 2, 1, 0, 0, 0),	1✔
449	DateTime::new_utc(2014, 3, 1, 0, 0, 0),	1✔
450	),
451	TimeInterval::new_unchecked(	1✔
452	DateTime::new_utc(2014, 1, 1, 0, 0, 0),	1✔
453	DateTime::new_utc(2014, 3, 1, 0, 0, 0),	1✔
454	),
455	TimeInterval::new_unchecked(	1✔
456	DateTime::new_utc(2014, 2, 1, 0, 0, 0),	1✔
457	DateTime::new_utc(2014, 3, 1, 0, 0, 0),	1✔
458	),
459	],
460	[	1✔
461	(	1✔
462	"id".to_string(),	1✔
463	FeatureData::NullableText(vec![	1✔
464	Some("S0".to_owned()),	1✔
465	Some("S1".to_owned()),	1✔
466	Some("S0".to_owned()),	1✔
467	None,	1✔
468	Some("S2".to_owned()),	1✔
469	]),	1✔
470	),	1✔
471	(	1✔
472	"value".to_string(),	1✔
473	FeatureData::NullableFloat(vec![	1✔
474	Some(0.),	1✔
475	Some(2.),	1✔
476	Some(1.),	1✔
477	Some(3.),	1✔
478	None,	1✔
479	]),	1✔
480	),	1✔
481	]	1✔
482	.iter()	1✔
483	.cloned()	1✔
484	.collect(),	1✔
485	CacheHint::default(),	1✔
486	)
487	.unwrap(),	1✔
488	)
489	.boxed();	1✔
490
491	let exe_ctc = MockExecutionContext::test_default();	1✔
492
493	let operator = FeatureAttributeValuesOverTime {	1✔
494	params: FeatureAttributeValuesOverTimeParams {	1✔
495	id_column: "id".to_owned(),	1✔
496	value_column: "value".to_owned(),	1✔
497	},	1✔
498	sources: point_source.into(),	1✔
499	};	1✔
500
501	let operator = operator	1✔
502	.boxed()	1✔
503	.initialize(WorkflowOperatorPath::initialize_root(), &exe_ctc)	1✔
504	.await	1✔
505	.unwrap();	1✔
506
507	let query_processor = operator.query_processor().unwrap().json_vega().unwrap();	1✔
508
509	let result = query_processor	1✔
510	.plot_query(	1✔
511	PlotQueryRectangle::new(	1✔
512	BoundingBox2D::new((-180., -90.).into(), (180., 90.).into()).unwrap(),	1✔
513	TimeInterval::default(),	1✔
514	PlotSeriesSelection::all(),	1✔
515	),	1✔
516	&exe_ctc.mock_query_context(ChunkByteSize::MIN),	1✔
517	)	1✔
518	.await	1✔
519	.unwrap();	1✔
520
521	assert!(matches!(result.metadata, PlotMetaData::None));	1✔
522
523	let vega_json: Value = serde_json::from_str(&result.vega_string).unwrap();	1✔
524
525	assert_eq!(	1✔
526	vega_json,	1✔
527	json!({	1✔
528	"$schema": "https://vega.github.io/schema/vega-lite/v4.17.0.json",	1✔
529	"data": {	1✔
530	"values": [{	1✔
531	"x": "2014-01-01T00:00:00+00:00",	1✔
532	"y": 0.0,	1✔
533	"series": "S0"	1✔
534	}, {	1✔
535	"x": "2014-02-01T00:00:00+00:00",	1✔
536	"y": 1.0,	1✔
537	"series": "S0"	1✔
538	}, {	1✔
539	"x": "2014-01-01T00:00:00+00:00",	1✔
540	"y": 2.0,	1✔
541	"series": "S1"	1✔
542	}]	1✔
543	},	1✔
544	"description": "Multi Line Chart",	1✔
545	"encoding": {	1✔
546	"x": {	1✔
547	"field": "x",	1✔
548	"title": "Time",	1✔
549	"type": "temporal"	1✔
550	},	1✔
551	"y": {	1✔
552	"field": "y",	1✔
553	"title": "",	1✔
554	"type": "quantitative"	1✔
555	},	1✔
556	"color": {	1✔
557	"field": "series",	1✔
558	"scale": {	1✔
559	"scheme": "category20"	1✔
560	}	1✔
561	}	1✔
562	},	1✔
563	"mark": {	1✔
564	"type": "line",	1✔
565	"line": true,	1✔
566	"point": true	1✔
567	}	1✔
568	})	1✔
569	);	1✔
570	}	1✔
571
572	#[tokio::test]
573	#[allow(clippy::too_many_lines)]
574	async fn plot_with_duplicates() {	1✔
575	let point_source = MockFeatureCollectionSource::single(	1✔
576	MultiPointCollection::from_data(	1✔
577	MultiPoint::many(vec![	1✔
578	vec![(-13.95, 20.05)],	1✔
579	vec![(-14.05, 20.05)],	1✔
580	vec![(-13.95, 20.05)],	1✔
581	vec![(-13.95, 20.05)],	1✔
582	])
583	.unwrap(),	1✔
584	vec![	1✔
585	TimeInterval::new_unchecked(	1✔
586	DateTime::new_utc(2014, 1, 1, 0, 0, 0),	1✔
587	DateTime::new_utc(2014, 2, 1, 0, 0, 0),	1✔
588	),
589	TimeInterval::new_unchecked(	1✔
590	DateTime::new_utc(2014, 1, 1, 0, 0, 0),	1✔
591	DateTime::new_utc(2014, 3, 1, 0, 0, 0),	1✔
592	),
593	TimeInterval::new_unchecked(	1✔
594	DateTime::new_utc(2014, 2, 1, 0, 0, 0),	1✔
595	DateTime::new_utc(2014, 3, 1, 0, 0, 0),	1✔
596	),
597	TimeInterval::new_unchecked(	1✔
598	DateTime::new_utc(2014, 2, 1, 0, 0, 0),	1✔
599	DateTime::new_utc(2014, 3, 1, 0, 0, 0),	1✔
600	),
601	],
602	[	1✔
603	(	1✔
604	"id".to_string(),	1✔
605	FeatureData::Text(vec![	1✔
606	"S0".to_owned(),	1✔
607	"S1".to_owned(),	1✔
608	"S0".to_owned(),	1✔
609	"S0".to_owned(),	1✔
610	]),	1✔
611	),	1✔
612	(	1✔
613	"value".to_string(),	1✔
614	FeatureData::Float(vec![0., 2., 1., 1.]),	1✔
615	),	1✔
616	]	1✔
617	.iter()	1✔
618	.cloned()	1✔
619	.collect(),	1✔
620	CacheHint::default(),	1✔
621	)
622	.unwrap(),	1✔
623	)
624	.boxed();	1✔
625
626	let exe_ctx = MockExecutionContext::test_default();	1✔
627
628	let operator = FeatureAttributeValuesOverTime {	1✔
629	params: FeatureAttributeValuesOverTimeParams {	1✔
630	id_column: "id".to_owned(),	1✔
631	value_column: "value".to_owned(),	1✔
632	},	1✔
633	sources: point_source.into(),	1✔
634	};	1✔
635
636	let operator = operator	1✔
637	.boxed()	1✔
638	.initialize(WorkflowOperatorPath::initialize_root(), &exe_ctx)	1✔
639	.await	1✔
640	.unwrap();	1✔
641
642	let query_processor = operator.query_processor().unwrap().json_vega().unwrap();	1✔
643
644	let result = query_processor	1✔
645	.plot_query(	1✔
646	PlotQueryRectangle::new(	1✔
647	BoundingBox2D::new((-180., -90.).into(), (180., 90.).into()).unwrap(),	1✔
648	TimeInterval::default(),	1✔
649	PlotSeriesSelection::all(),	1✔
650	),	1✔
651	&exe_ctx.mock_query_context(ChunkByteSize::MIN),	1✔
652	)	1✔
653	.await	1✔
654	.unwrap();	1✔
655
656	assert!(matches!(result.metadata, PlotMetaData::None));	1✔
657
658	let vega_json: Value = serde_json::from_str(&result.vega_string).unwrap();	1✔
659
660	assert_eq!(	1✔
661	vega_json,	1✔
662	json!({	1✔
663	"$schema": "https://vega.github.io/schema/vega-lite/v4.17.0.json",	1✔
664	"data": {	1✔
665	"values": [{	1✔
666	"x": "2014-01-01T00:00:00+00:00",	1✔
667	"y": 0.0,	1✔
668	"series": "S0"	1✔
669	}, {	1✔
670	"x": "2014-02-01T00:00:00+00:00",	1✔
671	"y": 1.0,	1✔
672	"series": "S0"	1✔
673	}, {	1✔
674	"x": "2014-02-01T00:00:00+00:00",	1✔
675	"y": 1.0,	1✔
676	"series": "S0"	1✔
677	}, {	1✔
678	"x": "2014-01-01T00:00:00+00:00",	1✔
679	"y": 2.0,	1✔
680	"series": "S1"	1✔
681	}]	1✔
682	},	1✔
683	"description": "Multi Line Chart",	1✔
684	"encoding": {	1✔
685	"x": {	1✔
686	"field": "x",	1✔
687	"title": "Time",	1✔
688	"type": "temporal"	1✔
689	},	1✔
690	"y": {	1✔
691	"field": "y",	1✔
692	"title": "",	1✔
693	"type": "quantitative"	1✔
694	},	1✔
695	"color": {	1✔
696	"field": "series",	1✔
697	"scale": {	1✔
698	"scheme": "category20"	1✔
699	}	1✔
700	}	1✔
701	},	1✔
702	"mark": {	1✔
703	"type": "line",	1✔
704	"line": true,	1✔
705	"point": true	1✔
706	}	1✔
707	})	1✔
708	);	1✔
709	}	1✔
710	}

geo-engine / geoengine / 19241805651

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