3929938005

Build Type

push

github

Committed by GitHub

Commit Message

Merge #713

Run Details

84930 of 96741 relevant lines covered (87.79%)

79640.1 hits per line

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

95.31

/operators/src/plot/temporal_vector_line_plot.rs

use crate::engine::{
    CreateSpan, ExecutionContext, InitializedPlotOperator, InitializedVectorOperator, Operator,
    OperatorName, PlotOperator, PlotQueryProcessor, PlotResultDescriptor, QueryContext,
    SingleVectorSource, TypedPlotQueryProcessor, VectorQueryProcessor,
};
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},
};
use tracing::{span, Level};

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>,
        context: &dyn ExecutionContext,
    ) -> Result<Box<dyn InitializedPlotOperator>> {
        let source = self.sources.vector.initialize(context).await?;
        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 {
            result_descriptor: in_desc.into(),
            vector_source: source,
            state: self.params,
        }
        .boxed())
    }

    span_fn!(FeatureAttributeValuesOverTime);
}

/// The initialization of `FeatureAttributeValuesOverTime`
pub struct InitializedFeatureAttributeValuesOverTime {
    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
    }
}

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

geo-engine / geoengine / 3929938005

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

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