12417045631

Committed 19 Dec 2024 04:45PM UTC coverage: 90.354% (-0.2%) from 90.512%

Build # 12417045631

Build Type

Pull #998

github

Committed by

web-flow

Commit Message

Merge 9e7b54661 into 34e12969f

Pull Request Pull Request #998: quota logging wip

Run Details

834 of 1211 new or added lines in 66 files covered. (68.87%)

227 existing lines in 20 files now uncovered.

133835 of 148123 relevant lines covered (90.35%)

54353.34 hits per line

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

95.39

/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::TryStreamExt;
use geoengine_datatypes::primitives::{FeatureDataType, PlotQueryRectangle};
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;
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()
    }
}

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

geo-engine / geoengine / 12417045631

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