13053993104

Committed 30 Jan 2025 01:35PM UTC coverage: 90.027% (-0.07%) from 90.093%

Build # 13053993104

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1011 from geo-engine/update-2025-01-17

update to rust 1.84

Run Details

46 of 54 new or added lines in 14 files covered. (85.19%)

102 existing lines in 41 files now uncovered.

125596 of 139510 relevant lines covered (90.03%)

57696.13 hits per line

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

95.15

/operators/src/plot/temporal_raster_mean_plot.rs

use crate::engine::{
    CanonicOperatorName, ExecutionContext, InitializedPlotOperator, InitializedRasterOperator,
    InitializedSources, Operator, OperatorName, PlotOperator, PlotQueryProcessor,
    PlotResultDescriptor, QueryContext, QueryProcessor, RasterQueryProcessor, SingleRasterSource,
    TypedPlotQueryProcessor, WorkflowOperatorPath,
};
use crate::util::math::average_floor;
use crate::util::Result;
use async_trait::async_trait;
use futures::stream::BoxStream;
use futures::StreamExt;
use geoengine_datatypes::plots::{AreaLineChart, Plot, PlotData};
use geoengine_datatypes::primitives::{
    BandSelection, Measurement, PlotQueryRectangle, RasterQueryRectangle, TimeInstance,
    TimeInterval,
};
use geoengine_datatypes::raster::{Pixel, RasterTile2D};
use serde::{Deserialize, Serialize};
use snafu::ensure;
use std::collections::BTreeMap;

pub const MEAN_RASTER_PIXEL_VALUES_OVER_TIME_NAME: &str = "Mean Raster Pixel Values over Time";

/// A plot that shows the mean values of rasters over time as an area plot.
pub type MeanRasterPixelValuesOverTime =
    Operator<MeanRasterPixelValuesOverTimeParams, SingleRasterSource>;

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

/// The parameter spec for `MeanRasterPixelValuesOverTime`
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct MeanRasterPixelValuesOverTimeParams {
    /// Where should the x-axis (time) tick be positioned?
    /// At either time start, time end or in the center.
    pub time_position: MeanRasterPixelValuesOverTimePosition,

    /// Whether to fill the area under the curve.
    #[serde(default = "default_true")]
    pub area: bool,
}

const fn default_true() -> bool {
    true
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum MeanRasterPixelValuesOverTimePosition {
    Start,
    Center,
    End,
}

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

        let initalized_sources = self
            .sources
            .initialize_sources(path.clone(), context)
            .await?;
        let raster = initalized_sources.raster;

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

        // TODO: implement multi-band functionality and remove this check
        ensure!(
            in_desc.bands.len() == 1,
            crate::error::OperatorDoesNotSupportMultiBandsSourcesYet {
                operator: MeanRasterPixelValuesOverTime::TYPE_NAME
            }
        );

        let initialized_operator = InitializedMeanRasterPixelValuesOverTime {
            name,
            result_descriptor: in_desc.into(),
            raster,
            state: self.params,
        };

        Ok(initialized_operator.boxed())
    }

    span_fn!(MeanRasterPixelValuesOverTime);
}

/// The initialization of `MeanRasterPixelValuesOverTime`
pub struct InitializedMeanRasterPixelValuesOverTime {
    name: CanonicOperatorName,
    result_descriptor: PlotResultDescriptor,
    raster: Box<dyn InitializedRasterOperator>,
    state: MeanRasterPixelValuesOverTimeParams,
}

impl InitializedPlotOperator for InitializedMeanRasterPixelValuesOverTime {
    fn query_processor(&self) -> Result<TypedPlotQueryProcessor> {
        let input_processor = self.raster.query_processor()?;
        let time_position = self.state.time_position;
        let measurement = self.raster.result_descriptor().bands[0].measurement.clone(); // TODO: adjust for multibands
        let draw_area = self.state.area;

        let processor = call_on_generic_raster_processor!(input_processor, raster => {
            MeanRasterPixelValuesOverTimeQueryProcessor { raster, time_position, measurement, draw_area }.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 `TemporalRasterMeanPlot` about its input.
pub struct MeanRasterPixelValuesOverTimeQueryProcessor<P: Pixel> {
    raster: Box<dyn RasterQueryProcessor<RasterType = P>>,
    time_position: MeanRasterPixelValuesOverTimePosition,
    measurement: Measurement,
    draw_area: bool,
}

#[async_trait]
impl<P: Pixel> PlotQueryProcessor for MeanRasterPixelValuesOverTimeQueryProcessor<P> {
    type OutputFormat = PlotData;

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

    async fn plot_query<'a>(
        &'a self,
        query: PlotQueryRectangle,
        ctx: &'a dyn QueryContext,
    ) -> Result<Self::OutputFormat> {
        let means = Self::calculate_means(
            self.raster
                .query(
                    RasterQueryRectangle::from_qrect_and_bands(&query, BandSelection::first()),
                    ctx,
                )
                .await?,
            self.time_position,
        )
        .await?;

        let plot = Self::generate_plot(means, self.measurement.clone(), self.draw_area)?;

        let plot_data = plot.to_vega_embeddable(false)?;

        Ok(plot_data)
    }
}

impl<P: Pixel> MeanRasterPixelValuesOverTimeQueryProcessor<P> {
    async fn calculate_means(
        mut tile_stream: BoxStream<'_, Result<RasterTile2D<P>>>,
        position: MeanRasterPixelValuesOverTimePosition,
    ) -> Result<BTreeMap<TimeInstance, MeanCalculator>> {
        let mut means: BTreeMap<TimeInstance, MeanCalculator> = BTreeMap::new();

        while let Some(tile) = tile_stream.next().await {
            let tile = tile?;

            match tile.grid_array {
                geoengine_datatypes::raster::GridOrEmpty::Grid(g) => {
                    let time = Self::time_interval_projection(tile.time, position);
                    let mean = means.entry(time).or_default();
                    mean.add(g.masked_element_deref_iterator());
                }
                geoengine_datatypes::raster::GridOrEmpty::Empty(_) => (),
            }
        }

        Ok(means)
    }

    #[inline]
    fn time_interval_projection(
        time_interval: TimeInterval,
        time_position: MeanRasterPixelValuesOverTimePosition,
    ) -> TimeInstance {
        match time_position {
            MeanRasterPixelValuesOverTimePosition::Start => time_interval.start(),
            MeanRasterPixelValuesOverTimePosition::Center => TimeInstance::from_millis_unchecked(
                average_floor(time_interval.start().inner(), time_interval.end().inner()),
            ),
            MeanRasterPixelValuesOverTimePosition::End => time_interval.end(),
        }
    }

    fn generate_plot(
        means: BTreeMap<TimeInstance, MeanCalculator>,
        measurement: Measurement,
        draw_area: bool,
    ) -> Result<AreaLineChart> {
        let mut timestamps = Vec::with_capacity(means.len());
        let mut values = Vec::with_capacity(means.len());

        for (timestamp, mean_calculator) in means {
            timestamps.push(timestamp);
            values.push(mean_calculator.mean());
        }

        AreaLineChart::new(timestamps, values, measurement, draw_area).map_err(Into::into)
    }
}

struct MeanCalculator {
    mean: f64,
    n: usize,
}

impl Default for MeanCalculator {
    fn default() -> Self {
        Self { mean: 0., n: 0 }
    }
}

impl MeanCalculator {
    #[inline]
    fn add<P: Pixel, I: Iterator<Item = Option<P>>>(&mut self, values: I) {
        values.flatten().for_each(|value| {
            self.add_single_value(value);
        });
    }

    #[inline]
    fn add_single_value<P: Pixel>(&mut self, value: P) {
        let value: f64 = value.as_();

        if value.is_nan() {
            return;
        }

        self.n += 1;
        let delta = value - self.mean;
        self.mean += delta / (self.n as f64);
    }

    #[inline]
    fn mean(&self) -> f64 {
        self.mean
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    use crate::{
        engine::{
            ChunkByteSize, MockExecutionContext, MockQueryContext, RasterBandDescriptors,
            RasterOperator, RasterResultDescriptor,
        },
        source::GdalSource,
    };
    use crate::{
        mock::{MockRasterSource, MockRasterSourceParams},
        source::GdalSourceParameters,
    };
    use geoengine_datatypes::primitives::{
        BoundingBox2D, CacheHint, Measurement, PlotSeriesSelection, SpatialResolution, TimeInterval,
    };
    use geoengine_datatypes::{dataset::NamedData, plots::PlotMetaData, primitives::DateTime};
    use geoengine_datatypes::{raster::TilingSpecification, spatial_reference::SpatialReference};
    use geoengine_datatypes::{
        raster::{Grid2D, RasterDataType, TileInformation},
        util::test::TestDefault,
    };
    use serde_json::{json, Value};

    #[test]
    fn serialization() {
        let temporal_raster_mean_plot = MeanRasterPixelValuesOverTime {
            params: MeanRasterPixelValuesOverTimeParams {
                time_position: MeanRasterPixelValuesOverTimePosition::Start,
                area: true,
            },
            sources: SingleRasterSource {
                raster: GdalSource {
                    params: GdalSourceParameters {
                        data: NamedData::with_system_name("test"),
                    },
                }
                .boxed(),
            },
        };

        let serialized = json!({
            "type": "MeanRasterPixelValuesOverTime",
            "params": {
                "timePosition": "start",
                "area": true,
            },
            "sources": {
                "raster": {
                    "type": "GdalSource",
                    "params": {
                        "data": "test"
                    }
                }
            },
            "vectorSources": [],
        })
        .to_string();

        serde_json::from_str::<Box<dyn PlotOperator>>(&serialized).unwrap();

        let deserialized: MeanRasterPixelValuesOverTime =
            serde_json::from_str(&serialized).unwrap();

        assert_eq!(deserialized.params, temporal_raster_mean_plot.params);
    }

    #[tokio::test]
    async fn single_raster() {
        let tile_size_in_pixels = [3, 2].into();
        let tiling_specification = TilingSpecification {
            origin_coordinate: [0.0, 0.0].into(),
            tile_size_in_pixels,
        };
        let execution_context = MockExecutionContext::new_with_tiling_spec(tiling_specification);

        let temporal_raster_mean_plot = MeanRasterPixelValuesOverTime {
            params: MeanRasterPixelValuesOverTimeParams {
                time_position: MeanRasterPixelValuesOverTimePosition::Center,
                area: true,
            },
            sources: SingleRasterSource {
                raster: generate_mock_raster_source(
                    vec![TimeInterval::new(
                        TimeInstance::from(DateTime::new_utc(1990, 1, 1, 0, 0, 0)),
                        TimeInstance::from(DateTime::new_utc(2000, 1, 1, 0, 0, 0)),
                    )
                    .unwrap()],
                    vec![vec![1, 2, 3, 4, 5, 6]],
                ),
            },
        };

        let temporal_raster_mean_plot = temporal_raster_mean_plot
            .boxed()
            .initialize(WorkflowOperatorPath::initialize_root(), &execution_context)
            .await
            .unwrap();

        let processor = temporal_raster_mean_plot
            .query_processor()
            .unwrap()
            .json_vega()
            .unwrap();

        let result = processor
            .plot_query(
                PlotQueryRectangle {
                    spatial_bounds: BoundingBox2D::new((-180., -90.).into(), (180., 90.).into())
                        .unwrap(),
                    time_interval: TimeInterval::default(),
                    spatial_resolution: SpatialResolution::one(),
                    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": "1995-01-01T00:00:00+00:00",
                        "y": 3.5
                    }]
                },
                "description": "Area Plot",
                "encoding": {
                    "x": {
                        "field": "x",
                        "title": "Time",
                        "type": "temporal"
                    },
                    "y": {
                        "field": "y",
                        "title": "",
                        "type": "quantitative"
                    }
                },
                "mark": {
                    "type": "area",
                    "line": true,
                    "point": true
                }
            })
        );
    }

    fn generate_mock_raster_source(
        time_intervals: Vec<TimeInterval>,
        values_vec: Vec<Vec<u8>>,
    ) -> Box<dyn RasterOperator> {
        assert_eq!(time_intervals.len(), values_vec.len());
        assert!(values_vec.iter().all(|v| v.len() == 6));

        let mut tiles = Vec::with_capacity(time_intervals.len());
        for (time_interval, values) in time_intervals.into_iter().zip(values_vec) {
            tiles.push(RasterTile2D::new_with_tile_info(
                time_interval,
                TileInformation {
                    global_geo_transform: TestDefault::test_default(),
                    global_tile_position: [0, 0].into(),
                    tile_size_in_pixels: [3, 2].into(),
                },
                0,
                Grid2D::new([3, 2].into(), values).unwrap().into(),
                CacheHint::default(),
            ));
        }

        MockRasterSource {
            params: MockRasterSourceParams {
                data: tiles,
                result_descriptor: RasterResultDescriptor {
                    data_type: RasterDataType::U8,
                    spatial_reference: SpatialReference::epsg_4326().into(),
                    time: None,
                    bbox: None,
                    resolution: None,
                    bands: RasterBandDescriptors::new_single_band(),
                },
            },
        }
        .boxed()
    }

    #[tokio::test]
    async fn raster_series() {
        let tile_size_in_pixels = [3, 2].into();
        let tiling_specification = TilingSpecification {
            origin_coordinate: [0.0, 0.0].into(),
            tile_size_in_pixels,
        };
        let execution_context = MockExecutionContext::new_with_tiling_spec(tiling_specification);

        let temporal_raster_mean_plot = MeanRasterPixelValuesOverTime {
            params: MeanRasterPixelValuesOverTimeParams {
                time_position: MeanRasterPixelValuesOverTimePosition::Start,
                area: true,
            },

            sources: SingleRasterSource {
                raster: generate_mock_raster_source(
                    vec![
                        TimeInterval::new(
                            TimeInstance::from(DateTime::new_utc(1990, 1, 1, 0, 0, 0)),
                            TimeInstance::from(DateTime::new_utc(1995, 1, 1, 0, 0, 0)),
                        )
                        .unwrap(),
                        TimeInterval::new(
                            TimeInstance::from(DateTime::new_utc(1995, 1, 1, 0, 0, 0)),
                            TimeInstance::from(DateTime::new_utc(2000, 1, 1, 0, 0, 0)),
                        )
                        .unwrap(),
                        TimeInterval::new(
                            TimeInstance::from(DateTime::new_utc(2000, 1, 1, 0, 0, 0)),
                            TimeInstance::from(DateTime::new_utc(2005, 1, 1, 0, 0, 0)),
                        )
                        .unwrap(),
                    ],
                    vec![
                        vec![1, 2, 3, 4, 5, 6],
                        vec![9, 9, 8, 8, 8, 9],
                        vec![3, 4, 5, 6, 7, 8],
                    ],
                ),
            },
        };

        let temporal_raster_mean_plot = temporal_raster_mean_plot
            .boxed()
            .initialize(WorkflowOperatorPath::initialize_root(), &execution_context)
            .await
            .unwrap();

        let processor = temporal_raster_mean_plot
            .query_processor()
            .unwrap()
            .json_vega()
            .unwrap();

        let result = processor
            .plot_query(
                PlotQueryRectangle {
                    spatial_bounds: BoundingBox2D::new((-180., -90.).into(), (180., 90.).into())
                        .unwrap(),
                    time_interval: TimeInterval::default(),
                    spatial_resolution: SpatialResolution::one(),
                    attributes: PlotSeriesSelection::all(),
                },
                &MockQueryContext::new(ChunkByteSize::MIN),
            )
            .await
            .unwrap();

        assert_eq!(
            result,
            AreaLineChart::new(
                vec![
                    TimeInstance::from(DateTime::new_utc(1990, 1, 1, 0, 0, 0)),
                    TimeInstance::from(DateTime::new_utc(1995, 1, 1, 0, 0, 0)),
                    TimeInstance::from(DateTime::new_utc(2000, 1, 1, 0, 0, 0))
                ],
                vec![3.5, 8.5, 5.5],
                Measurement::Unitless,
                true,
            )
            .unwrap()
            .to_vega_embeddable(false)
            .unwrap()
        );
    }
}

1	use crate::engine::{
2	CanonicOperatorName, ExecutionContext, InitializedPlotOperator, InitializedRasterOperator,
3	InitializedSources, Operator, OperatorName, PlotOperator, PlotQueryProcessor,
4	PlotResultDescriptor, QueryContext, QueryProcessor, RasterQueryProcessor, SingleRasterSource,
5	TypedPlotQueryProcessor, WorkflowOperatorPath,
6	};
7	use crate::util::math::average_floor;
8	use crate::util::Result;
9	use async_trait::async_trait;
10	use futures::stream::BoxStream;
11	use futures::StreamExt;
12	use geoengine_datatypes::plots::{AreaLineChart, Plot, PlotData};
13	use geoengine_datatypes::primitives::{
14	BandSelection, Measurement, PlotQueryRectangle, RasterQueryRectangle, TimeInstance,
15	TimeInterval,
16	};
17	use geoengine_datatypes::raster::{Pixel, RasterTile2D};
18	use serde::{Deserialize, Serialize};
19	use snafu::ensure;
20	use std::collections::BTreeMap;
21
22	pub const MEAN_RASTER_PIXEL_VALUES_OVER_TIME_NAME: &str = "Mean Raster Pixel Values over Time";
23
24	/// A plot that shows the mean values of rasters over time as an area plot.
25	pub type MeanRasterPixelValuesOverTime =
26	Operator<MeanRasterPixelValuesOverTimeParams, SingleRasterSource>;
27
28	impl OperatorName for MeanRasterPixelValuesOverTime {
29	const TYPE_NAME: &'static str = "MeanRasterPixelValuesOverTime";
30	}
31
32	/// The parameter spec for `MeanRasterPixelValuesOverTime`
33	#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
34	#[serde(rename_all = "camelCase")]
35	pub struct MeanRasterPixelValuesOverTimeParams {
36	/// Where should the x-axis (time) tick be positioned?
37	/// At either time start, time end or in the center.
38	pub time_position: MeanRasterPixelValuesOverTimePosition,
39
40	/// Whether to fill the area under the curve.
41	#[serde(default = "default_true")]
42	pub area: bool,
43	}
44
45	const fn default_true() -> bool {	×
46	true	×
47	}	×
48
49	#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
50	#[serde(rename_all = "camelCase")]
51	pub enum MeanRasterPixelValuesOverTimePosition {
52	Start,
53	Center,
54	End,
55	}
56
UNCOV 57	#[typetag::serde]	×
58	#[async_trait]
59	impl PlotOperator for MeanRasterPixelValuesOverTime {
60	async fn _initialize(
61	self: Box<Self>,
62	path: WorkflowOperatorPath,
63	context: &dyn ExecutionContext,
64	) -> Result<Box<dyn InitializedPlotOperator>> {	3✔
65	let name = CanonicOperatorName::from(&self);	3✔
66
67	let initalized_sources = self	3✔
68	.sources	3✔
69	.initialize_sources(path.clone(), context)	3✔
70	.await?;	3✔
71	let raster = initalized_sources.raster;	3✔
72		3✔
73	let in_desc = raster.result_descriptor().clone();	3✔
74		3✔
75	// TODO: implement multi-band functionality and remove this check	3✔
76	ensure!(	3✔
77	in_desc.bands.len() == 1,	3✔
78	crate::error::OperatorDoesNotSupportMultiBandsSourcesYet {	×
79	operator: MeanRasterPixelValuesOverTime::TYPE_NAME	×
80	}	×
81	);
82
83	let initialized_operator = InitializedMeanRasterPixelValuesOverTime {	3✔
84	name,	3✔
85	result_descriptor: in_desc.into(),	3✔
86	raster,	3✔
87	state: self.params,	3✔
88	};	3✔
89		3✔
90	Ok(initialized_operator.boxed())	3✔
91	}	6✔
92
93	span_fn!(MeanRasterPixelValuesOverTime);
94	}
95
96	/// The initialization of `MeanRasterPixelValuesOverTime`
97	pub struct InitializedMeanRasterPixelValuesOverTime {
98	name: CanonicOperatorName,
99	result_descriptor: PlotResultDescriptor,
100	raster: Box<dyn InitializedRasterOperator>,
101	state: MeanRasterPixelValuesOverTimeParams,
102	}
103
104	impl InitializedPlotOperator for InitializedMeanRasterPixelValuesOverTime {
105	fn query_processor(&self) -> Result<TypedPlotQueryProcessor> {	3✔
106	let input_processor = self.raster.query_processor()?;	3✔
107	let time_position = self.state.time_position;	3✔
108	let measurement = self.raster.result_descriptor().bands[0].measurement.clone(); // TODO: adjust for multibands	3✔
109	let draw_area = self.state.area;	3✔
110
111	let processor = call_on_generic_raster_processor!(input_processor, raster => {	3✔
112	MeanRasterPixelValuesOverTimeQueryProcessor { raster, time_position, measurement, draw_area }.boxed()	2✔
113	});
114
115	Ok(TypedPlotQueryProcessor::JsonVega(processor))	3✔
116	}	3✔
117
118	fn result_descriptor(&self) -> &PlotResultDescriptor {	×
119	&self.result_descriptor	×
120	}	×
121
122	fn canonic_name(&self) -> CanonicOperatorName {	×
123	self.name.clone()	×
124	}	×
125	}
126
127	/// A query processor that calculates the `TemporalRasterMeanPlot` about its input.
128	pub struct MeanRasterPixelValuesOverTimeQueryProcessor<P: Pixel> {
129	raster: Box<dyn RasterQueryProcessor<RasterType = P>>,
130	time_position: MeanRasterPixelValuesOverTimePosition,
131	measurement: Measurement,
132	draw_area: bool,
133	}
134
135	#[async_trait]
136	impl<P: Pixel> PlotQueryProcessor for MeanRasterPixelValuesOverTimeQueryProcessor<P> {
137	type OutputFormat = PlotData;
138
139	fn plot_type(&self) -> &'static str {	×
140	MEAN_RASTER_PIXEL_VALUES_OVER_TIME_NAME	×
141	}	×
142
143	async fn plot_query<'a>(
144	&'a self,
145	query: PlotQueryRectangle,
146	ctx: &'a dyn QueryContext,
147	) -> Result<Self::OutputFormat> {	3✔
148	let means = Self::calculate_means(	3✔
149	self.raster	3✔
150	.query(	3✔
151	RasterQueryRectangle::from_qrect_and_bands(&query, BandSelection::first()),	3✔
152	ctx,	3✔
153	)	3✔
154	.await?,	3✔
155	self.time_position,	3✔
156	)	3✔
157	.await?;	3✔
158
159	let plot = Self::generate_plot(means, self.measurement.clone(), self.draw_area)?;	3✔
160
161	let plot_data = plot.to_vega_embeddable(false)?;	3✔
162
163	Ok(plot_data)	3✔
164	}	6✔
165	}
166
167	impl<P: Pixel> MeanRasterPixelValuesOverTimeQueryProcessor<P> {
168	async fn calculate_means(	3✔
169	mut tile_stream: BoxStream<'_, Result<RasterTile2D<P>>>,	3✔
170	position: MeanRasterPixelValuesOverTimePosition,	3✔
171	) -> Result<BTreeMap<TimeInstance, MeanCalculator>> {	3✔
172	let mut means: BTreeMap<TimeInstance, MeanCalculator> = BTreeMap::new();	3✔
173
174	while let Some(tile) = tile_stream.next().await {	88,351✔
175	let tile = tile?;	88,348✔
176
177	match tile.grid_array {	88,348✔
178	geoengine_datatypes::raster::GridOrEmpty::Grid(g) => {	12✔
179	let time = Self::time_interval_projection(tile.time, position);	12✔
180	let mean = means.entry(time).or_default();	12✔
181	mean.add(g.masked_element_deref_iterator());	12✔
182	}	12✔
183	geoengine_datatypes::raster::GridOrEmpty::Empty(_) => (),	88,336✔
184	}
185	}
186
187	Ok(means)	3✔
188	}	3✔
189
190	#[inline]
191	fn time_interval_projection(	12✔
192	time_interval: TimeInterval,	12✔
193	time_position: MeanRasterPixelValuesOverTimePosition,	12✔
194	) -> TimeInstance {	12✔
195	match time_position {	12✔
196	MeanRasterPixelValuesOverTimePosition::Start => time_interval.start(),	11✔
197	MeanRasterPixelValuesOverTimePosition::Center => TimeInstance::from_millis_unchecked(	1✔
198	average_floor(time_interval.start().inner(), time_interval.end().inner()),	1✔
199	),	1✔
200	MeanRasterPixelValuesOverTimePosition::End => time_interval.end(),	×
201	}
202	}	12✔
203
204	fn generate_plot(	3✔
205	means: BTreeMap<TimeInstance, MeanCalculator>,	3✔
206	measurement: Measurement,	3✔
207	draw_area: bool,	3✔
208	) -> Result<AreaLineChart> {	3✔
209	let mut timestamps = Vec::with_capacity(means.len());	3✔
210	let mut values = Vec::with_capacity(means.len());	3✔
211
212	for (timestamp, mean_calculator) in means {	9✔
213	timestamps.push(timestamp);	6✔
214	values.push(mean_calculator.mean());	6✔
215	}	6✔
216
217	AreaLineChart::new(timestamps, values, measurement, draw_area).map_err(Into::into)	3✔
218	}	3✔
219	}
220
221	struct MeanCalculator {
222	mean: f64,
223	n: usize,
224	}
225
226	impl Default for MeanCalculator {
227	fn default() -> Self {	6✔
228	Self { mean: 0., n: 0 }	6✔
229	}	6✔
230	}
231
232	impl MeanCalculator {
233	#[inline]
234	fn add<P: Pixel, I: Iterator<Item = Option<P>>>(&mut self, values: I) {	12✔
235	values.flatten().for_each(\|value\| {	5,024✔
236	self.add_single_value(value);	5,024✔
237	});	5,024✔
238	}	12✔
239
240	#[inline]
241	fn add_single_value<P: Pixel>(&mut self, value: P) {	5,024✔
242	let value: f64 = value.as_();	5,024✔
243		5,024✔
244	if value.is_nan() {	5,024✔
245	return;	×
246	}	5,024✔
247		5,024✔
248	self.n += 1;	5,024✔
249	let delta = value - self.mean;	5,024✔
250	self.mean += delta / (self.n as f64);	5,024✔
251	}	5,024✔
252
253	#[inline]
254	fn mean(&self) -> f64 {	6✔
255	self.mean	6✔
256	}	6✔
257	}
258
259	#[cfg(test)]
260	mod tests {
261	use super::*;
262
263	use crate::{
264	engine::{
265	ChunkByteSize, MockExecutionContext, MockQueryContext, RasterBandDescriptors,
266	RasterOperator, RasterResultDescriptor,
267	},
268	source::GdalSource,
269	};
270	use crate::{
271	mock::{MockRasterSource, MockRasterSourceParams},
272	source::GdalSourceParameters,
273	};
274	use geoengine_datatypes::primitives::{
275	BoundingBox2D, CacheHint, Measurement, PlotSeriesSelection, SpatialResolution, TimeInterval,
276	};
277	use geoengine_datatypes::{dataset::NamedData, plots::PlotMetaData, primitives::DateTime};
278	use geoengine_datatypes::{raster::TilingSpecification, spatial_reference::SpatialReference};
279	use geoengine_datatypes::{
280	raster::{Grid2D, RasterDataType, TileInformation},
281	util::test::TestDefault,
282	};
283	use serde_json::{json, Value};
284
285	#[test]
286	fn serialization() {	1✔
287	let temporal_raster_mean_plot = MeanRasterPixelValuesOverTime {	1✔
288	params: MeanRasterPixelValuesOverTimeParams {	1✔
289	time_position: MeanRasterPixelValuesOverTimePosition::Start,	1✔
290	area: true,	1✔
291	},	1✔
292	sources: SingleRasterSource {	1✔
293	raster: GdalSource {	1✔
294	params: GdalSourceParameters {	1✔
295	data: NamedData::with_system_name("test"),	1✔
296	},	1✔
297	}	1✔
298	.boxed(),	1✔
299	},	1✔
300	};	1✔
301		1✔
302	let serialized = json!({	1✔
303	"type": "MeanRasterPixelValuesOverTime",	1✔
304	"params": {	1✔
305	"timePosition": "start",	1✔
306	"area": true,	1✔
307	},	1✔
308	"sources": {	1✔
309	"raster": {	1✔
310	"type": "GdalSource",	1✔
311	"params": {	1✔
312	"data": "test"	1✔
313	}	1✔
314	}	1✔
315	},	1✔
316	"vectorSources": [],	1✔
317	})	1✔
318	.to_string();	1✔
319		1✔
320	serde_json::from_str::<Box<dyn PlotOperator>>(&serialized).unwrap();	1✔
321		1✔
322	let deserialized: MeanRasterPixelValuesOverTime =	1✔
323	serde_json::from_str(&serialized).unwrap();	1✔
324		1✔
325	assert_eq!(deserialized.params, temporal_raster_mean_plot.params);	1✔
326	}	1✔
327
328	#[tokio::test]
329	async fn single_raster() {	1✔
330	let tile_size_in_pixels = [3, 2].into();	1✔
331	let tiling_specification = TilingSpecification {	1✔
332	origin_coordinate: [0.0, 0.0].into(),	1✔
333	tile_size_in_pixels,	1✔
334	};	1✔
335	let execution_context = MockExecutionContext::new_with_tiling_spec(tiling_specification);	1✔
336		1✔
337	let temporal_raster_mean_plot = MeanRasterPixelValuesOverTime {	1✔
338	params: MeanRasterPixelValuesOverTimeParams {	1✔
339	time_position: MeanRasterPixelValuesOverTimePosition::Center,	1✔
340	area: true,	1✔
341	},	1✔
342	sources: SingleRasterSource {	1✔
343	raster: generate_mock_raster_source(	1✔
344	vec![TimeInterval::new(	1✔
345	TimeInstance::from(DateTime::new_utc(1990, 1, 1, 0, 0, 0)),	1✔
346	TimeInstance::from(DateTime::new_utc(2000, 1, 1, 0, 0, 0)),	1✔
347	)	1✔
348	.unwrap()],	1✔
349	vec![vec![1, 2, 3, 4, 5, 6]],	1✔
350	),	1✔
351	},	1✔
352	};	1✔
353		1✔
354	let temporal_raster_mean_plot = temporal_raster_mean_plot	1✔
355	.boxed()	1✔
356	.initialize(WorkflowOperatorPath::initialize_root(), &execution_context)	1✔
357	.await	1✔
358	.unwrap();	1✔
359		1✔
360	let processor = temporal_raster_mean_plot	1✔
361	.query_processor()	1✔
362	.unwrap()	1✔
363	.json_vega()	1✔
364	.unwrap();	1✔
365		1✔
366	let result = processor	1✔
367	.plot_query(	1✔
368	PlotQueryRectangle {	1✔
369	spatial_bounds: BoundingBox2D::new((-180., -90.).into(), (180., 90.).into())	1✔
370	.unwrap(),	1✔
371	time_interval: TimeInterval::default(),	1✔
372	spatial_resolution: SpatialResolution::one(),	1✔
373	attributes: PlotSeriesSelection::all(),	1✔
374	},	1✔
375	&MockQueryContext::new(ChunkByteSize::MIN),	1✔
376	)	1✔
377	.await	1✔
378	.unwrap();	1✔
379		1✔
380	assert!(matches!(result.metadata, PlotMetaData::None));	1✔
381		1✔
382	let vega_json: Value = serde_json::from_str(&result.vega_string).unwrap();	1✔
383		1✔
384	assert_eq!(	1✔
385	vega_json,	1✔
386	json!({	1✔
387	"$schema": "https://vega.github.io/schema/vega-lite/v4.17.0.json",	1✔
388	"data": {	1✔
389	"values": [{	1✔
390	"x": "1995-01-01T00:00:00+00:00",	1✔
391	"y": 3.5	1✔
392	}]	1✔
393	},	1✔
394	"description": "Area Plot",	1✔
395	"encoding": {	1✔
396	"x": {	1✔
397	"field": "x",	1✔
398	"title": "Time",	1✔
399	"type": "temporal"	1✔
400	},	1✔
401	"y": {	1✔
402	"field": "y",	1✔
403	"title": "",	1✔
404	"type": "quantitative"	1✔
405	}	1✔
406	},	1✔
407	"mark": {	1✔
408	"type": "area",	1✔
409	"line": true,	1✔
410	"point": true	1✔
411	}	1✔
412	})	1✔
413	);	1✔
414	}	1✔
415
416	fn generate_mock_raster_source(	2✔
417	time_intervals: Vec<TimeInterval>,	2✔
418	values_vec: Vec<Vec<u8>>,	2✔
419	) -> Box<dyn RasterOperator> {	2✔
420	assert_eq!(time_intervals.len(), values_vec.len());	2✔
421	assert!(values_vec.iter().all(\|v\| v.len() == 6));	4✔
422
423	let mut tiles = Vec::with_capacity(time_intervals.len());	2✔
424	for (time_interval, values) in time_intervals.into_iter().zip(values_vec) {	4✔
425	tiles.push(RasterTile2D::new_with_tile_info(	4✔
426	time_interval,	4✔
427	TileInformation {	4✔
428	global_geo_transform: TestDefault::test_default(),	4✔
429	global_tile_position: [0, 0].into(),	4✔
430	tile_size_in_pixels: [3, 2].into(),	4✔
431	},	4✔
432	0,	4✔
433	Grid2D::new([3, 2].into(), values).unwrap().into(),	4✔
434	CacheHint::default(),	4✔
435	));	4✔
436	}	4✔
437
438	MockRasterSource {	2✔
439	params: MockRasterSourceParams {	2✔
440	data: tiles,	2✔
441	result_descriptor: RasterResultDescriptor {	2✔
442	data_type: RasterDataType::U8,	2✔
443	spatial_reference: SpatialReference::epsg_4326().into(),	2✔
444	time: None,	2✔
445	bbox: None,	2✔
446	resolution: None,	2✔
447	bands: RasterBandDescriptors::new_single_band(),	2✔
448	},	2✔
449	},	2✔
450	}	2✔
451	.boxed()	2✔
452	}	2✔
453
454	#[tokio::test]
455	async fn raster_series() {	1✔
456	let tile_size_in_pixels = [3, 2].into();	1✔
457	let tiling_specification = TilingSpecification {	1✔
458	origin_coordinate: [0.0, 0.0].into(),	1✔
459	tile_size_in_pixels,	1✔
460	};	1✔
461	let execution_context = MockExecutionContext::new_with_tiling_spec(tiling_specification);	1✔
462		1✔
463	let temporal_raster_mean_plot = MeanRasterPixelValuesOverTime {	1✔
464	params: MeanRasterPixelValuesOverTimeParams {	1✔
465	time_position: MeanRasterPixelValuesOverTimePosition::Start,	1✔
466	area: true,	1✔
467	},	1✔
468		1✔
469	sources: SingleRasterSource {	1✔
470	raster: generate_mock_raster_source(	1✔
471	vec![	1✔
472	TimeInterval::new(	1✔
473	TimeInstance::from(DateTime::new_utc(1990, 1, 1, 0, 0, 0)),	1✔
474	TimeInstance::from(DateTime::new_utc(1995, 1, 1, 0, 0, 0)),	1✔
475	)	1✔
476	.unwrap(),	1✔
477	TimeInterval::new(	1✔
478	TimeInstance::from(DateTime::new_utc(1995, 1, 1, 0, 0, 0)),	1✔
479	TimeInstance::from(DateTime::new_utc(2000, 1, 1, 0, 0, 0)),	1✔
480	)	1✔
481	.unwrap(),	1✔
482	TimeInterval::new(	1✔
483	TimeInstance::from(DateTime::new_utc(2000, 1, 1, 0, 0, 0)),	1✔
484	TimeInstance::from(DateTime::new_utc(2005, 1, 1, 0, 0, 0)),	1✔
485	)	1✔
486	.unwrap(),	1✔
487	],	1✔
488	vec![	1✔
489	vec![1, 2, 3, 4, 5, 6],	1✔
490	vec![9, 9, 8, 8, 8, 9],	1✔
491	vec![3, 4, 5, 6, 7, 8],	1✔
492	],	1✔
493	),	1✔
494	},	1✔
495	};	1✔
496		1✔
497	let temporal_raster_mean_plot = temporal_raster_mean_plot	1✔
498	.boxed()	1✔
499	.initialize(WorkflowOperatorPath::initialize_root(), &execution_context)	1✔
500	.await	1✔
501	.unwrap();	1✔
502		1✔
503	let processor = temporal_raster_mean_plot	1✔
504	.query_processor()	1✔
505	.unwrap()	1✔
506	.json_vega()	1✔
507	.unwrap();	1✔
508		1✔
509	let result = processor	1✔
510	.plot_query(	1✔
511	PlotQueryRectangle {	1✔
512	spatial_bounds: BoundingBox2D::new((-180., -90.).into(), (180., 90.).into())	1✔
513	.unwrap(),	1✔
514	time_interval: TimeInterval::default(),	1✔
515	spatial_resolution: SpatialResolution::one(),	1✔
516	attributes: PlotSeriesSelection::all(),	1✔
517	},	1✔
518	&MockQueryContext::new(ChunkByteSize::MIN),	1✔
519	)	1✔
520	.await	1✔
521	.unwrap();	1✔
522		1✔
523	assert_eq!(	1✔
524	result,	1✔
525	AreaLineChart::new(	1✔
526	vec![	1✔
527	TimeInstance::from(DateTime::new_utc(1990, 1, 1, 0, 0, 0)),	1✔
528	TimeInstance::from(DateTime::new_utc(1995, 1, 1, 0, 0, 0)),	1✔
529	TimeInstance::from(DateTime::new_utc(2000, 1, 1, 0, 0, 0))	1✔
530	],	1✔
531	vec![3.5, 8.5, 5.5],	1✔
532	Measurement::Unitless,	1✔
533	true,	1✔
534	)	1✔
535	.unwrap()	1✔
536	.to_vega_embeddable(false)	1✔
537	.unwrap()	1✔
538	);	1✔
539	}	1✔
540	}

geo-engine / geoengine / 13053993104

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