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

94.35

/operators/src/plot/temporal_vector_line_plot.rs

use crate::engine::{
    CanonicOperatorName, ExecutionContext, InitializedPlotOperator, InitializedSources,
    InitializedVectorOperator, Operator, OperatorName, PlotOperator, PlotQueryProcessor,
    PlotResultDescriptor, QueryContext, SingleVectorSource, TypedPlotQueryProcessor,
    VectorQueryProcessor, WorkflowOperatorPath,
};
use crate::engine::{QueryProcessor, VectorColumnInfo};
use crate::error;
use crate::util::Result;
use async_trait::async_trait;
use futures::StreamExt;
use geoengine_datatypes::primitives::{FeatureDataType, VectorQueryRectangle};
use geoengine_datatypes::{
    collections::FeatureCollection,
    plots::{Plot, PlotData},
};
use geoengine_datatypes::{
    collections::FeatureCollectionInfos,
    plots::{DataPoint, MultiLineChart},
};
use geoengine_datatypes::{
    primitives::{Geometry, Measurement, TimeInterval},
    util::arrow::ArrowTyped,
};
use serde::{Deserialize, Serialize};
use snafu::{ensure, ResultExt};
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, 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()
    }
}

/// 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: VectorQueryRectangle,
        ctx: &'a dyn QueryContext,
    ) -> Result<Self::OutputFormat> {
        let values = FeatureAttributeValues::<MAX_FEATURES>::default();

        let values = self
            .features
            .query(query, ctx)
            .await?
            .fold(Ok(values), |acc, features| async {
                match (acc, features) {
                    (Ok(mut acc), Ok(features)) => {
                        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)
                    }
                    (Err(err), _) | (_, Err(err)) => Err(err),
                }
            })
            .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)
            .context(error::DataType)
    }
}

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 geoengine_datatypes::util::test::TestDefault;
    use geoengine_datatypes::{
        collections::MultiPointCollection,
        plots::PlotMetaData,
        primitives::{
            BoundingBox2D, DateTime, FeatureData, MultiPoint, SpatialResolution, TimeInterval,
        },
    };
    use serde_json::{json, Value};

    use crate::{
        engine::{ChunkByteSize, MockExecutionContext, MockQueryContext, VectorOperator},
        mock::MockFeatureCollectionSource,
    };

    #[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(),
            )
            .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(
                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();

        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(),
            )
            .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(
                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();

        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(),
            )
            .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(
                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();

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

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