10178074589

Committed 31 Jul 2024 09:34AM UTC coverage: 91.068% (+0.4%) from 90.682%

Build # 10178074589

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #973 from geo-engine/remove-XGB-update-toolchain

Remove-XGB-update-toolchain

Run Details

81 of 88 new or added lines in 29 files covered. (92.05%)

456 existing lines in 119 files now uncovered.

131088 of 143945 relevant lines covered (91.07%)

53581.03 hits per line

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

97.11

/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, 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: 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)
            .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::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, 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`
UNCOV 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		3✔
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 {	3✔
70	column: self.params.id_column.clone()	×
71	}	×
72	);	3✔
73		3✔
74	ensure!(	3✔
75	columns.contains_key(&self.params.value_column),	3✔
76	error::ColumnDoesNotExist {	3✔
77	column: self.params.value_column.clone()	×
78	}	×
79	);	3✔
80		3✔
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		3✔
90	// TODO: ensure column is really an id	3✔
91	ensure!(	3✔
92	id_type == FeatureDataType::Text	3✔
93	\|\| id_type == FeatureDataType::Int	3✔
94	\|\| id_type == FeatureDataType::Category,	3✔
95	error::InvalidFeatureDataType,	3✔
96	);	3✔
97		3✔
98	ensure!(	3✔
99	value_type.is_numeric() \|\| value_type == FeatureDataType::Category,	3✔
100	error::InvalidFeatureDataType,	3✔
101	);	3✔
102		3✔
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	}	3✔
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✔
UNCOV 130	FeatureAttributeValuesOverTimeQueryProcessor { params: self.state.clone(), features }.boxed()	×
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>(
166	&'a self,
167	query: PlotQueryRectangle,
168	ctx: &'a dyn QueryContext,
169	) -> Result<Self::OutputFormat> {	3✔
170	let values = FeatureAttributeValues::<MAX_FEATURES>::default();	3✔
171		3✔
172	let values = self	3✔
173	.features	3✔
174	.query(query.into(), ctx)	3✔
175	.await?	3✔
176	.try_fold(values, \|mut acc, features\| async move {	3✔
177	let ids = features.data(&self.params.id_column)?;	3✔
178	let values = features.data(&self.params.value_column)?;	3✔
179		3✔
180	for ((id, value), &time) in ids	12✔
181	.strings_iter()	3✔
182	.zip(values.float_options_iter())	3✔
183	.zip(features.time_intervals())	3✔
184	{	3✔
185	if id.is_empty() \|\| value.is_none() {	12✔
186	continue;	3✔
187	}	10✔
188		10✔
189	let value = value.expect("checked above");	10✔
190		10✔
191	acc.add(id, (time, value));	10✔
192	}	3✔
193		3✔
194	Ok(acc)	3✔
195	})	3✔
196	.await?;	3✔
197		3✔
198	let data_points = values.get_data_points();	3✔
199	let measurement = Measurement::Unitless; // TODO: attach actual unit if we know it	3✔
200	MultiLineChart::new(data_points, measurement)	3✔
201	.to_vega_embeddable(false)	3✔
202	.context(error::DataType)	3✔
203	}	3✔
204	}
205
206	struct TemporalValue {
207	pub time: TimeInterval,
208	pub value: f64,
209	}
210
211	impl From<(TimeInterval, f64)> for TemporalValue {
212	fn from(value: (TimeInterval, f64)) -> Self {	10✔
213	Self {	10✔
214	time: value.0,	10✔
215	value: value.1,	10✔
216	}	10✔
217	}	10✔
218	}
219
220	struct FeatureAttributeValues<const LENGTH: usize> {
221	values: HashMap<String, Vec<TemporalValue>>,
222	}
223
224	impl<const LENGTH: usize> Default for FeatureAttributeValues<LENGTH> {
225	fn default() -> Self {	3✔
226	Self {	3✔
227	values: HashMap::with_capacity(LENGTH),	3✔
228	}	3✔
229	}	3✔
230	}
231
232	impl<const LENGTH: usize> FeatureAttributeValues<LENGTH> {
233	/// Add value to the data structure. If `id` is new and there are already `LENGTH` existing
234	/// `id`-entries, the value is ignored
235	pub fn add<V>(&mut self, id: String, value: V)	10✔
236	where	10✔
237	V: Into<TemporalValue>,	10✔
238	{	10✔
239	let len = self.values.len();	10✔
240		10✔
241	match self.values.entry(id) {	10✔
242	Occupied(mut entry) => entry.get_mut().push(value.into()),	4✔
243	Vacant(entry) => {	6✔
244	if len < LENGTH {	6✔
245	entry.insert(vec![value.into()]);	6✔
246	}	6✔
247	}
248	}
249	}	10✔
250
251	pub fn get_data_points(mut self) -> Vec<DataPoint> {	3✔
252	let mut data = self	3✔
253	.values	3✔
254	.drain()	3✔
255	.flat_map(\|(id, values)\| {	6✔
256	values.into_iter().map(move \|value\| DataPoint {	10✔
257	series: id.clone(),	10✔
258	time: value.time.start(),	10✔
259	value: value.value,	10✔
260	})	10✔
261	})	6✔
262	.collect::<Vec<_>>();	3✔
263		3✔
264	data.sort_unstable_by(\|a, b\| match a.series.cmp(&b.series) {	11✔
265	Ordering::Equal => a.time.cmp(&b.time),	4✔
266	other => other,	7✔
267	});	11✔
268	data	3✔
269	}	3✔
270	}
271
272	#[cfg(test)]
273	mod tests {
274	use super::*;
275	use geoengine_datatypes::primitives::{CacheHint, PlotSeriesSelection};
276	use geoengine_datatypes::util::test::TestDefault;
277	use geoengine_datatypes::{
278	collections::MultiPointCollection,
279	plots::PlotMetaData,
280	primitives::{
281	BoundingBox2D, DateTime, FeatureData, MultiPoint, SpatialResolution, TimeInterval,
282	},
283	};
284	use serde_json::{json, Value};
285
286	use crate::{
287	engine::{ChunkByteSize, MockExecutionContext, MockQueryContext, VectorOperator},
288	mock::MockFeatureCollectionSource,
289	};
290
291	#[tokio::test]
292	#[allow(clippy::too_many_lines)]
293	async fn plot() {	1✔
294	let point_source = MockFeatureCollectionSource::single(	1✔
295	MultiPointCollection::from_data(	1✔
296	MultiPoint::many(vec![	1✔
297	vec![(-13.95, 20.05)],	1✔
298	vec![(-14.05, 20.05)],	1✔
299	vec![(-13.95, 20.05)],	1✔
300	])	1✔
301	.unwrap(),	1✔
302	vec![	1✔
303	TimeInterval::new_unchecked(	1✔
304	DateTime::new_utc(2014, 1, 1, 0, 0, 0),	1✔
305	DateTime::new_utc(2014, 2, 1, 0, 0, 0),	1✔
306	),	1✔
307	TimeInterval::new_unchecked(	1✔
308	DateTime::new_utc(2014, 1, 1, 0, 0, 0),	1✔
309	DateTime::new_utc(2014, 3, 1, 0, 0, 0),	1✔
310	),	1✔
311	TimeInterval::new_unchecked(	1✔
312	DateTime::new_utc(2014, 2, 1, 0, 0, 0),	1✔
313	DateTime::new_utc(2014, 3, 1, 0, 0, 0),	1✔
314	),	1✔
315	],	1✔
316	[	1✔
317	(	1✔
318	"id".to_string(),	1✔
319	FeatureData::Text(vec!["S0".to_owned(), "S1".to_owned(), "S0".to_owned()]),	1✔
320	),	1✔
321	("value".to_string(), FeatureData::Float(vec![0., 2., 1.])),	1✔
322	]	1✔
323	.iter()	1✔
324	.cloned()	1✔
325	.collect(),	1✔
326	CacheHint::default(),	1✔
327	)	1✔
328	.unwrap(),	1✔
329	)	1✔
330	.boxed();	1✔
331		1✔
332	let exe_ctc = MockExecutionContext::test_default();	1✔
333		1✔
334	let operator = FeatureAttributeValuesOverTime {	1✔
335	params: FeatureAttributeValuesOverTimeParams {	1✔
336	id_column: "id".to_owned(),	1✔
337	value_column: "value".to_owned(),	1✔
338	},	1✔
339	sources: point_source.into(),	1✔
340	};	1✔
341		1✔
342	let operator = operator	1✔
343	.boxed()	1✔
344	.initialize(WorkflowOperatorPath::initialize_root(), &exe_ctc)	1✔
345	.await	1✔
346	.unwrap();	1✔
347		1✔
348	let query_processor = operator.query_processor().unwrap().json_vega().unwrap();	1✔
349		1✔
350	let result = query_processor	1✔
351	.plot_query(	1✔
352	PlotQueryRectangle {	1✔
353	spatial_bounds: BoundingBox2D::new((-180., -90.).into(), (180., 90.).into())	1✔
354	.unwrap(),	1✔
355	time_interval: TimeInterval::default(),	1✔
356	spatial_resolution: SpatialResolution::new(0.1, 0.1).unwrap(),	1✔
357	attributes: PlotSeriesSelection::all(),	1✔
358	},	1✔
359	&MockQueryContext::new(ChunkByteSize::MIN),	1✔
360	)	1✔
361	.await	1✔
362	.unwrap();	1✔
363		1✔
364	assert!(matches!(result.metadata, PlotMetaData::None));	1✔
365		1✔
366	let vega_json: Value = serde_json::from_str(&result.vega_string).unwrap();	1✔
367		1✔
368	assert_eq!(	1✔
369	vega_json,	1✔
370	json!({	1✔
371	"$schema": "https://vega.github.io/schema/vega-lite/v4.17.0.json",	1✔
372	"data": {	1✔
373	"values": [{	1✔
374	"x": "2014-01-01T00:00:00+00:00",	1✔
375	"y": 0.0,	1✔
376	"series": "S0"	1✔
377	}, {	1✔
378	"x": "2014-02-01T00:00:00+00:00",	1✔
379	"y": 1.0,	1✔
380	"series": "S0"	1✔
381	}, {	1✔
382	"x": "2014-01-01T00:00:00+00:00",	1✔
383	"y": 2.0,	1✔
384	"series": "S1"	1✔
385	}]	1✔
386	},	1✔
387	"description": "Multi Line Chart",	1✔
388	"encoding": {	1✔
389	"x": {	1✔
390	"field": "x",	1✔
391	"title": "Time",	1✔
392	"type": "temporal"	1✔
393	},	1✔
394	"y": {	1✔
395	"field": "y",	1✔
396	"title": "",	1✔
397	"type": "quantitative"	1✔
398	},	1✔
399	"color": {	1✔
400	"field": "series",	1✔
401	"scale": {	1✔
402	"scheme": "category20"	1✔
403	}	1✔
404	}	1✔
405	},	1✔
406	"mark": {	1✔
407	"type": "line",	1✔
408	"line": true,	1✔
409	"point": true	1✔
410	}	1✔
411	})	1✔
412	);	1✔
413	}	1✔
414
415	#[tokio::test]
416	#[allow(clippy::too_many_lines)]
417	async fn plot_with_nulls() {	1✔
418	let point_source = MockFeatureCollectionSource::single(	1✔
419	MultiPointCollection::from_data(	1✔
420	MultiPoint::many(vec![	1✔
421	vec![(-13.95, 20.05)],	1✔
422	vec![(-14.05, 20.05)],	1✔
423	vec![(-13.95, 20.05)],	1✔
424	vec![(-14.05, 20.05)],	1✔
425	vec![(-13.95, 20.05)],	1✔
426	])	1✔
427	.unwrap(),	1✔
428	vec![	1✔
429	TimeInterval::new_unchecked(	1✔
430	DateTime::new_utc(2014, 1, 1, 0, 0, 0),	1✔
431	DateTime::new_utc(2014, 2, 1, 0, 0, 0),	1✔
432	),	1✔
433	TimeInterval::new_unchecked(	1✔
434	DateTime::new_utc(2014, 1, 1, 0, 0, 0),	1✔
435	DateTime::new_utc(2014, 3, 1, 0, 0, 0),	1✔
436	),	1✔
437	TimeInterval::new_unchecked(	1✔
438	DateTime::new_utc(2014, 2, 1, 0, 0, 0),	1✔
439	DateTime::new_utc(2014, 3, 1, 0, 0, 0),	1✔
440	),	1✔
441	TimeInterval::new_unchecked(	1✔
442	DateTime::new_utc(2014, 1, 1, 0, 0, 0),	1✔
443	DateTime::new_utc(2014, 3, 1, 0, 0, 0),	1✔
444	),	1✔
445	TimeInterval::new_unchecked(	1✔
446	DateTime::new_utc(2014, 2, 1, 0, 0, 0),	1✔
447	DateTime::new_utc(2014, 3, 1, 0, 0, 0),	1✔
448	),	1✔
449	],	1✔
450	[	1✔
451	(	1✔
452	"id".to_string(),	1✔
453	FeatureData::NullableText(vec![	1✔
454	Some("S0".to_owned()),	1✔
455	Some("S1".to_owned()),	1✔
456	Some("S0".to_owned()),	1✔
457	None,	1✔
458	Some("S2".to_owned()),	1✔
459	]),	1✔
460	),	1✔
461	(	1✔
462	"value".to_string(),	1✔
463	FeatureData::NullableFloat(vec![	1✔
464	Some(0.),	1✔
465	Some(2.),	1✔
466	Some(1.),	1✔
467	Some(3.),	1✔
468	None,	1✔
469	]),	1✔
470	),	1✔
471	]	1✔
472	.iter()	1✔
473	.cloned()	1✔
474	.collect(),	1✔
475	CacheHint::default(),	1✔
476	)	1✔
477	.unwrap(),	1✔
478	)	1✔
479	.boxed();	1✔
480		1✔
481	let exe_ctc = MockExecutionContext::test_default();	1✔
482		1✔
483	let operator = FeatureAttributeValuesOverTime {	1✔
484	params: FeatureAttributeValuesOverTimeParams {	1✔
485	id_column: "id".to_owned(),	1✔
486	value_column: "value".to_owned(),	1✔
487	},	1✔
488	sources: point_source.into(),	1✔
489	};	1✔
490		1✔
491	let operator = operator	1✔
492	.boxed()	1✔
493	.initialize(WorkflowOperatorPath::initialize_root(), &exe_ctc)	1✔
494	.await	1✔
495	.unwrap();	1✔
496		1✔
497	let query_processor = operator.query_processor().unwrap().json_vega().unwrap();	1✔
498		1✔
499	let result = query_processor	1✔
500	.plot_query(	1✔
501	PlotQueryRectangle {	1✔
502	spatial_bounds: BoundingBox2D::new((-180., -90.).into(), (180., 90.).into())	1✔
503	.unwrap(),	1✔
504	time_interval: TimeInterval::default(),	1✔
505	spatial_resolution: SpatialResolution::new(0.1, 0.1).unwrap(),	1✔
506	attributes: PlotSeriesSelection::all(),	1✔
507	},	1✔
508	&MockQueryContext::new(ChunkByteSize::MIN),	1✔
509	)	1✔
510	.await	1✔
511	.unwrap();	1✔
512		1✔
513	assert!(matches!(result.metadata, PlotMetaData::None));	1✔
514		1✔
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	}	1✔
563
564	#[tokio::test]
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	CacheHint::default(),	1✔
613	)	1✔
614	.unwrap(),	1✔
615	)	1✔
616	.boxed();	1✔
617		1✔
618	let exe_ctc = MockExecutionContext::test_default();	1✔
619		1✔
620	let operator = FeatureAttributeValuesOverTime {	1✔
621	params: FeatureAttributeValuesOverTimeParams {	1✔
622	id_column: "id".to_owned(),	1✔
623	value_column: "value".to_owned(),	1✔
624	},	1✔
625	sources: point_source.into(),	1✔
626	};	1✔
627		1✔
628	let operator = operator	1✔
629	.boxed()	1✔
630	.initialize(WorkflowOperatorPath::initialize_root(), &exe_ctc)	1✔
631	.await	1✔
632	.unwrap();	1✔
633		1✔
634	let query_processor = operator.query_processor().unwrap().json_vega().unwrap();	1✔
635		1✔
636	let result = query_processor	1✔
637	.plot_query(	1✔
638	PlotQueryRectangle {	1✔
639	spatial_bounds: BoundingBox2D::new((-180., -90.).into(), (180., 90.).into())	1✔
640	.unwrap(),	1✔
641	time_interval: TimeInterval::default(),	1✔
642	spatial_resolution: SpatialResolution::new(0.1, 0.1).unwrap(),	1✔
643	attributes: PlotSeriesSelection::all(),	1✔
644	},	1✔
645	&MockQueryContext::new(ChunkByteSize::MIN),	1✔
646	)	1✔
647	.await	1✔
648	.unwrap();	1✔
649		1✔
650	assert!(matches!(result.metadata, PlotMetaData::None));	1✔
651		1✔
652	let vega_json: Value = serde_json::from_str(&result.vega_string).unwrap();	1✔
653		1✔
654	assert_eq!(	1✔
655	vega_json,	1✔
656	json!({	1✔
657	"$schema": "https://vega.github.io/schema/vega-lite/v4.17.0.json",	1✔
658	"data": {	1✔
659	"values": [{	1✔
660	"x": "2014-01-01T00:00:00+00:00",	1✔
661	"y": 0.0,	1✔
662	"series": "S0"	1✔
663	}, {	1✔
664	"x": "2014-02-01T00:00:00+00:00",	1✔
665	"y": 1.0,	1✔
666	"series": "S0"	1✔
667	}, {	1✔
668	"x": "2014-02-01T00:00:00+00:00",	1✔
669	"y": 1.0,	1✔
670	"series": "S0"	1✔
671	}, {	1✔
672	"x": "2014-01-01T00:00:00+00:00",	1✔
673	"y": 2.0,	1✔
674	"series": "S1"	1✔
675	}]	1✔
676	},	1✔
677	"description": "Multi Line Chart",	1✔
678	"encoding": {	1✔
679	"x": {	1✔
680	"field": "x",	1✔
681	"title": "Time",	1✔
682	"type": "temporal"	1✔
683	},	1✔
684	"y": {	1✔
685	"field": "y",	1✔
686	"title": "",	1✔
687	"type": "quantitative"	1✔
688	},	1✔
689	"color": {	1✔
690	"field": "series",	1✔
691	"scale": {	1✔
692	"scheme": "category20"	1✔
693	}	1✔
694	}	1✔
695	},	1✔
696	"mark": {	1✔
697	"type": "line",	1✔
698	"line": true,	1✔
699	"point": true	1✔
700	}	1✔
701	})	1✔
702	);	1✔
703	}	1✔
704	}

geo-engine / geoengine / 10178074589

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