19438548963

Committed 17 Nov 2025 05:27PM UTC coverage: 88.512%. First build

Build # 19438548963

Build Type

Pull #1083

github

Committed by

web-flow

Commit Message

Merge c3ef76798 into f0e6b6470

Pull Request Pull Request #1083: feat: new gdal source workflow optimization

Coverage Stats

6329 of 7650 new or added lines in 78 files covered. (82.73%)

116305 of 131400 relevant lines covered (88.51%)

494085.33 hits per line

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

95.16

/operators/src/processing/time_shift.rs

use crate::engine::{
    CanonicOperatorName, ExecutionContext, InitializedRasterOperator,
    InitializedSingleRasterOrVectorOperator, InitializedSources, InitializedVectorOperator,
    Operator, OperatorName, QueryContext, QueryProcessor, RasterOperator, RasterQueryProcessor,
    RasterResultDescriptor, ResultDescriptor, SingleRasterOrVectorSource,
    TypedRasterQueryProcessor, TypedVectorQueryProcessor, VectorOperator, VectorQueryProcessor,
    VectorResultDescriptor, WorkflowOperatorPath,
};
use crate::optimization::OptimizationError;
use crate::util::Result;
use crate::util::input::RasterOrVectorOperator;
use async_trait::async_trait;
use futures::StreamExt;
use futures::stream::BoxStream;
use geoengine_datatypes::collections::{
    FeatureCollection, FeatureCollectionInfos, FeatureCollectionModifications,
};
use geoengine_datatypes::error::{BoxedResultExt, ErrorSource};
use geoengine_datatypes::primitives::{
    ColumnSelection, Duration, Geometry, RasterQueryRectangle, SpatialResolution, TimeGranularity,
    TimeInstance, TimeInterval,
};
use geoengine_datatypes::primitives::{TimeStep, VectorQueryRectangle};
use geoengine_datatypes::raster::{Pixel, RasterTile2D};
use geoengine_datatypes::util::arrow::ArrowTyped;
use serde::{Deserialize, Serialize};
use snafu::Snafu;

/// Project the query rectangle to a new time interval.
pub type TimeShift = Operator<TimeShiftParams, SingleRasterOrVectorSource>;

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

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "camelCase")]
pub enum TimeShiftParams {
    /// Shift the query rectangle relative with a time step
    #[serde(rename_all = "camelCase")]
    Relative {
        granularity: TimeGranularity,
        value: i32,
    },
    /// Set the time interval to a fixed value
    #[serde(rename_all = "camelCase")]
    Absolute { time_interval: TimeInterval },
}

pub trait TimeShiftOperation: Send + Sync + Copy {
    type State: Send + Sync + Copy;

    fn shift(
        &self,
        time_interval: TimeInterval,
    ) -> Result<(TimeInterval, Self::State), TimeShiftError>;

    fn reverse_shift(
        &self,
        time_interval: TimeInterval,
        state: Self::State,
    ) -> Result<TimeInterval, TimeShiftError>;
}

#[derive(Debug, Clone, Copy)]
pub struct RelativeForwardShift {
    step: TimeStep,
}

impl TimeShiftOperation for RelativeForwardShift {
    type State = ();

    fn shift(
        &self,
        time_interval: TimeInterval,
    ) -> Result<(TimeInterval, Self::State), TimeShiftError> {
        let time_interval = time_interval + self.step;
        let time_interval = time_interval.boxed_context(error::TimeOverflow)?;
        Ok((time_interval, ()))
    }

    fn reverse_shift(
        &self,
        time_interval: TimeInterval,
        _state: Self::State,
    ) -> Result<TimeInterval, TimeShiftError> {
        let reversed_time_interval = time_interval - self.step;
        reversed_time_interval.boxed_context(error::TimeOverflow)
    }
}

#[derive(Debug, Clone, Copy)]
pub struct RelativeBackwardShift {
    step: TimeStep,
}

impl TimeShiftOperation for RelativeBackwardShift {
    type State = ();

    fn shift(
        &self,
        time_interval: TimeInterval,
    ) -> Result<(TimeInterval, Self::State), TimeShiftError> {
        let time_interval = time_interval - self.step;
        let time_interval = time_interval.boxed_context(error::TimeOverflow)?;
        Ok((time_interval, ()))
    }

    fn reverse_shift(
        &self,
        time_interval: TimeInterval,
        _state: Self::State,
    ) -> Result<TimeInterval, TimeShiftError> {
        let reversed_time_interval = time_interval + self.step;
        reversed_time_interval.boxed_context(error::TimeOverflow)
    }
}

#[derive(Debug, Clone, Copy)]
pub struct AbsoluteShift {
    time_interval: TimeInterval,
}

impl TimeShiftOperation for AbsoluteShift {
    type State = (Duration, Duration);

    fn shift(
        &self,
        time_interval: TimeInterval,
    ) -> Result<(TimeInterval, Self::State), TimeShiftError> {
        let time_start_difference = time_interval.start() - self.time_interval.start();
        let time_end_difference = time_interval.end() - self.time_interval.end();

        Ok((
            self.time_interval,
            (time_start_difference, time_end_difference),
        ))
    }

    fn reverse_shift(
        &self,
        time_interval: TimeInterval,
        (time_start_difference, time_end_difference): Self::State,
    ) -> Result<TimeInterval, TimeShiftError> {
        let t1 = time_interval.start() + time_start_difference.num_milliseconds();
        let t2 = time_interval.end() + time_end_difference.num_milliseconds();
        TimeInterval::new(t1, t2).boxed_context(error::FaultyTimeInterval { t1, t2 })
    }
}

#[derive(Debug, Snafu)]
#[snafu(visibility(pub(crate)), context(suffix(false)), module(error))]
pub enum TimeShiftError {
    #[snafu(display("Output type must match the type of the source"))]
    UnmatchedOutput,
    #[snafu(display("Shifting the time led to an overflowing time interval"))]
    TimeOverflow { source: Box<dyn ErrorSource> },
    #[snafu(display("Shifting the time to a faulty time interval: {t1} / {t2}"))]
    FaultyTimeInterval {
        source: Box<dyn ErrorSource>,
        t1: TimeInstance,
        t2: TimeInstance,
    },
    #[snafu(display("Modifying the timestamps of the feature collection failed"))]
    FeatureCollectionTimeModification { source: Box<dyn ErrorSource> },
}

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

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

        match (init_sources.source, self.params.clone()) {
            (
                InitializedSingleRasterOrVectorOperator::Vector(source),
                TimeShiftParams::Relative { granularity, value },
            ) if value.is_positive() => {
                let shift = RelativeForwardShift {
                    step: TimeStep {
                        granularity,
                        step: value.unsigned_abs(),
                    },
                };

                let result_descriptor = shift_result_descriptor(source.result_descriptor(), shift);

                Ok(Box::new(InitializedVectorTimeShift {
                    name,
                    path,
                    params: self.params.clone(),
                    source,
                    result_descriptor,
                    shift,
                }))
            }
            (
                InitializedSingleRasterOrVectorOperator::Vector(source),
                TimeShiftParams::Relative { granularity, value },
            ) => {
                let shift = RelativeBackwardShift {
                    step: TimeStep {
                        granularity,
                        step: value.unsigned_abs(),
                    },
                };

                let result_descriptor = shift_result_descriptor(source.result_descriptor(), shift);

                Ok(Box::new(InitializedVectorTimeShift {
                    name,
                    path,
                    params: self.params.clone(),
                    source,
                    result_descriptor,
                    shift,
                }))
            }
            (
                InitializedSingleRasterOrVectorOperator::Vector(source),
                TimeShiftParams::Absolute { time_interval },
            ) => {
                let shift = AbsoluteShift { time_interval };

                let result_descriptor = shift_result_descriptor(source.result_descriptor(), shift);

                Ok(Box::new(InitializedVectorTimeShift {
                    name,
                    path,
                    params: self.params.clone(),
                    source,
                    result_descriptor,
                    shift,
                }))
            }
            (InitializedSingleRasterOrVectorOperator::Raster(_), _) => {
                Err(TimeShiftError::UnmatchedOutput.into())
            }
        }
    }

    span_fn!(TimeShift);
}

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

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

        match (init_sources.source, self.params.clone()) {
            (
                InitializedSingleRasterOrVectorOperator::Raster(source),
                TimeShiftParams::Relative { granularity, value },
            ) if value.is_positive() => {
                let shift = RelativeForwardShift {
                    step: TimeStep {
                        granularity,
                        step: value.unsigned_abs(),
                    },
                };

                let result_descriptor = shift_result_descriptor(source.result_descriptor(), shift);

                Ok(Box::new(InitializedRasterTimeShift {
                    name,
                    path,
                    params: self.params.clone(),
                    source,
                    result_descriptor,
                    shift,
                }))
            }
            (
                InitializedSingleRasterOrVectorOperator::Raster(source),
                TimeShiftParams::Relative { granularity, value },
            ) => {
                let shift = RelativeBackwardShift {
                    step: TimeStep {
                        granularity,
                        step: value.unsigned_abs(),
                    },
                };

                let result_descriptor = shift_result_descriptor(source.result_descriptor(), shift);

                Ok(Box::new(InitializedRasterTimeShift {
                    name,
                    path,
                    params: self.params.clone(),
                    source,
                    result_descriptor,
                    shift,
                }))
            }
            (
                InitializedSingleRasterOrVectorOperator::Raster(source),
                TimeShiftParams::Absolute { time_interval },
            ) => {
                let shift = AbsoluteShift { time_interval };

                let result_descriptor = shift_result_descriptor(source.result_descriptor(), shift);

                Ok(Box::new(InitializedRasterTimeShift {
                    name,
                    path,
                    params: self.params.clone(),
                    source,
                    result_descriptor,
                    shift,
                }))
            }
            (InitializedSingleRasterOrVectorOperator::Vector(_), _) => {
                Err(TimeShiftError::UnmatchedOutput.into())
            }
        }
    }

    span_fn!(TimeShift);
}

fn shift_result_descriptor<R: ResultDescriptor, S: TimeShiftOperation>(
    result_descriptor: &R,
    shift: S,
) -> R {
    result_descriptor.map_time(|time| {
        if let Some(time) = time {
            shift.shift(*time).map(|r| r.0).ok()
        } else {
            None
        }
    })
}

pub struct InitializedVectorTimeShift<Shift: TimeShiftOperation> {
    name: CanonicOperatorName,
    path: WorkflowOperatorPath,
    params: TimeShiftParams,
    source: Box<dyn InitializedVectorOperator>,
    result_descriptor: VectorResultDescriptor,
    shift: Shift,
}

pub struct InitializedRasterTimeShift<Shift: TimeShiftOperation> {
    name: CanonicOperatorName,
    path: WorkflowOperatorPath,
    params: TimeShiftParams,
    source: Box<dyn InitializedRasterOperator>,
    result_descriptor: RasterResultDescriptor,
    shift: Shift,
}

impl<Shift: TimeShiftOperation + 'static> InitializedVectorOperator
    for InitializedVectorTimeShift<Shift>
{
    fn result_descriptor(&self) -> &VectorResultDescriptor {
        &self.result_descriptor
    }

    fn query_processor(&self) -> Result<TypedVectorQueryProcessor> {
        let source_processor = self.source.query_processor()?;

        Ok(
            call_on_generic_vector_processor!(source_processor, processor => VectorTimeShiftProcessor {
                processor,
                result_descriptor: self.result_descriptor.clone(),
                shift: self.shift,
            }.boxed().into()),
        )
    }

    fn canonic_name(&self) -> CanonicOperatorName {
        self.name.clone()
    }

    fn name(&self) -> &'static str {
        TimeShift::TYPE_NAME
    }

    fn path(&self) -> WorkflowOperatorPath {
        self.path.clone()
    }

    fn optimize(
        &self,
        target_resolution: SpatialResolution,
    ) -> Result<Box<dyn VectorOperator>, OptimizationError> {
        Ok(Box::new(TimeShift {
            params: self.params.clone(),
            sources: SingleRasterOrVectorSource {
                source: RasterOrVectorOperator::Vector(self.source.optimize(target_resolution)?),
            },
        }))
    }
}

impl<Shift: TimeShiftOperation + 'static> InitializedRasterOperator
    for InitializedRasterTimeShift<Shift>
{
    fn result_descriptor(&self) -> &RasterResultDescriptor {
        &self.result_descriptor
    }

    fn query_processor(&self) -> Result<TypedRasterQueryProcessor> {
        let source_processor = self.source.query_processor()?;

        Ok(
            call_on_generic_raster_processor!(source_processor, processor => RasterTimeShiftProcessor {
                processor,
                result_descriptor: self.result_descriptor.clone(),
                shift: self.shift,
            }.boxed().into()),
        )
    }

    fn canonic_name(&self) -> CanonicOperatorName {
        self.name.clone()
    }

    fn name(&self) -> &'static str {
        TimeShift::TYPE_NAME
    }

    fn path(&self) -> WorkflowOperatorPath {
        self.path.clone()
    }

    fn optimize(
        &self,
        target_resolution: SpatialResolution,
    ) -> Result<Box<dyn RasterOperator>, OptimizationError> {
        Ok(Box::new(TimeShift {
            params: self.params.clone(),
            sources: SingleRasterOrVectorSource {
                source: RasterOrVectorOperator::Raster(self.source.optimize(target_resolution)?),
            },
        }))
    }
}

pub struct RasterTimeShiftProcessor<Q, P, Shift: TimeShiftOperation>
where
    Q: RasterQueryProcessor<RasterType = P>,
{
    processor: Q,
    result_descriptor: RasterResultDescriptor,
    shift: Shift,
}

pub struct VectorTimeShiftProcessor<Q, G, Shift: TimeShiftOperation>
where
    G: Geometry,
    Q: VectorQueryProcessor<VectorType = FeatureCollection<G>>,
{
    processor: Q,
    result_descriptor: VectorResultDescriptor,
    shift: Shift,
}

#[async_trait]
impl<Q, G, Shift> VectorQueryProcessor for VectorTimeShiftProcessor<Q, G, Shift>
where
    G: Geometry + ArrowTyped + 'static,
    Q: VectorQueryProcessor<VectorType = FeatureCollection<G>>,
    Shift: TimeShiftOperation + 'static,
{
    type VectorType = FeatureCollection<G>;

    async fn vector_query<'a>(
        &'a self,
        query: VectorQueryRectangle,
        ctx: &'a dyn QueryContext,
    ) -> Result<BoxStream<'a, Result<Self::VectorType>>> {
        let (time_interval, state) = self.shift.shift(query.time_interval())?;

        let query = VectorQueryRectangle::new(
            query.spatial_bounds(),
            time_interval,
            ColumnSelection::all(),
        );
        let stream = self.processor.vector_query(query, ctx).await?;

        let stream = stream.then(move |collection| async move {
            let collection = collection?;
            let shift = self.shift;

            crate::util::spawn_blocking(move || {
                let time_intervals = collection
                    .time_intervals()
                    .iter()
                    .map(move |time| shift.reverse_shift(*time, state))
                    .collect::<Result<Vec<TimeInterval>, TimeShiftError>>()?;

                collection
                    .replace_time(&time_intervals)
                    .boxed_context(error::FeatureCollectionTimeModification)
                    .map_err(Into::into)
            })
            .await?
        });

        Ok(stream.boxed())
    }

    fn vector_result_descriptor(&self) -> &VectorResultDescriptor {
        &self.result_descriptor
    }
}

#[async_trait]
impl<Q, P, Shift> QueryProcessor for RasterTimeShiftProcessor<Q, P, Shift>
where
    Q: RasterQueryProcessor<RasterType = P>,
    P: Pixel,
    Shift: TimeShiftOperation,
{
    type Output = RasterTile2D<P>;
    type SpatialBounds = Q::SpatialBounds;
    type ResultDescription = RasterResultDescriptor;
    type Selection = Q::Selection;

    async fn _query<'a>(
        &'a self,
        query: RasterQueryRectangle,
        ctx: &'a dyn QueryContext,
    ) -> Result<BoxStream<'a, Result<Self::Output>>> {
        let (time_interval, state) = self.shift.shift(query.time_interval())?;
        let query = query.select_time_interval(time_interval);
        let stream = self.processor.raster_query(query, ctx).await?;

        let stream = stream.map(move |raster| {
            // reverse time shift for results
            let mut raster = raster?;

            raster.time = self.shift.reverse_shift(raster.time, state)?;

            Ok(raster)
        });

        Ok(Box::pin(stream))
    }

    fn result_descriptor(&self) -> &RasterResultDescriptor {
        &self.result_descriptor
    }
}

#[async_trait]
impl<Q, P, Shift> RasterQueryProcessor for RasterTimeShiftProcessor<Q, P, Shift>
where
    P: Pixel,
    Q: RasterQueryProcessor<RasterType = P>,
    Shift: TimeShiftOperation,
{
    type RasterType = P;

    async fn _time_query<'a>(
        &'a self,
        query: TimeInterval,
        ctx: &'a dyn QueryContext,
    ) -> Result<BoxStream<'a, Result<TimeInterval>>> {
        let (time_interval, state) = self.shift.shift(query)?;
        let stream = self.processor.time_query(time_interval, ctx).await?;

        let stream = stream
            .map(move |ti| {
                // reverse time shift for results
                ti.and_then(|t| self.shift.reverse_shift(t, state).map_err(Into::into)) // TODO: maybe we need a time_query error...
            })
            .boxed();

        Ok(stream)
    }
}

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

    use crate::{
        engine::{
            MockExecutionContext, MultipleRasterSources, RasterBandDescriptors, SingleRasterSource,
            SpatialGridDescriptor, TimeDescriptor,
        },
        mock::{MockFeatureCollectionSource, MockRasterSource, MockRasterSourceParams},
        processing::{Expression, ExpressionParams, RasterStacker, RasterStackerParams},
        source::{GdalSource, GdalSourceParameters},
        util::{gdal::add_ndvi_dataset, input::RasterOrVectorOperator},
    };
    use futures::StreamExt;
    use geoengine_datatypes::{
        collections::{ChunksEqualIgnoringCacheHint, MultiPointCollection},
        dataset::NamedData,
        primitives::{
            BandSelection, BoundingBox2D, CacheHint, Coordinate2D, DateTime, MultiPoint,
            TimeGranularity,
        },
        raster::{
            BoundedGrid, EmptyGrid2D, GeoTransform, GridBoundingBox2D, GridOrEmpty, GridShape2D,
            RasterDataType, RenameBands, TileInformation, TilingSpecification,
        },
        spatial_reference::SpatialReference,
        util::test::TestDefault,
    };

    #[test]
    fn test_ser_de_absolute() {
        let time_shift = TimeShift {
            sources: SingleRasterOrVectorSource {
                source: RasterOrVectorOperator::Raster(
                    GdalSource {
                        params: GdalSourceParameters::new(NamedData::with_system_name(
                            "test-raster",
                        )),
                    }
                    .boxed(),
                ),
            },
            params: TimeShiftParams::Absolute {
                time_interval: TimeInterval::new_unchecked(
                    DateTime::new_utc(2011, 1, 1, 0, 0, 0),
                    DateTime::new_utc(2012, 1, 1, 0, 0, 0),
                ),
            },
        };

        let serialized = serde_json::to_value(&time_shift).unwrap();

        assert_eq!(
            serialized,
            serde_json::json!({
                "params": {
                    "type": "absolute",
                    "timeInterval": {
                        "start": 1_293_840_000_000_i64,
                        "end": 1_325_376_000_000_i64
                    }
                },
                "sources": {
                    "source": {
                        "type": "GdalSource",
                        "params": {
                            "data": "test-raster",
                            "overviewLevel": null
                        }
                    }
                }
            })
        );

        let deserialized: TimeShift = serde_json::from_value(serialized).unwrap();

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

    #[test]
    fn test_ser_de_relative() {
        let time_shift = TimeShift {
            sources: SingleRasterOrVectorSource {
                source: RasterOrVectorOperator::Raster(
                    GdalSource {
                        params: GdalSourceParameters::new(NamedData::with_system_name(
                            "test-raster",
                        )),
                    }
                    .boxed(),
                ),
            },
            params: TimeShiftParams::Relative {
                granularity: TimeGranularity::Years,
                value: 1,
            },
        };

        let serialized = serde_json::to_value(&time_shift).unwrap();

        assert_eq!(
            serialized,
            serde_json::json!({
                "params": {
                    "type": "relative",
                    "granularity": "years",
                    "value": 1
                },
                "sources": {
                    "source": {
                        "type": "GdalSource",
                        "params": {
                            "data": "test-raster",
                            "overviewLevel": null
                        }
                    }
                }
            })
        );

        let deserialized: TimeShift = serde_json::from_value(serialized).unwrap();

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

    #[tokio::test]
    async fn test_absolute_vector_shift() {
        let execution_context = MockExecutionContext::test_default();
        let query_context = execution_context.mock_query_context_test_default();

        let source = MockFeatureCollectionSource::single(
            MultiPointCollection::from_data(
                MultiPoint::many(vec![(0., 0.), (1., 1.), (2., 2.)]).unwrap(),
                vec![
                    TimeInterval::new(
                        DateTime::new_utc(2009, 1, 1, 0, 0, 0),
                        DateTime::new_utc_with_millis(2010, 12, 31, 23, 59, 59, 999),
                    )
                    .unwrap(),
                    TimeInterval::new(
                        DateTime::new_utc(2009, 6, 3, 0, 0, 0),
                        DateTime::new_utc(2010, 7, 14, 0, 0, 0),
                    )
                    .unwrap(),
                    TimeInterval::new(
                        DateTime::new_utc(2010, 1, 1, 0, 0, 0),
                        DateTime::new_utc_with_millis(2011, 3, 31, 23, 59, 59, 999),
                    )
                    .unwrap(),
                ],
                Default::default(),
                CacheHint::default(),
            )
            .unwrap(),
        );

        let time_shift = TimeShift {
            sources: SingleRasterOrVectorSource {
                source: RasterOrVectorOperator::Vector(source.boxed()),
            },
            params: TimeShiftParams::Absolute {
                time_interval: TimeInterval::new(
                    DateTime::new_utc(2009, 1, 1, 0, 0, 0),
                    DateTime::new_utc(2009, 6, 1, 0, 0, 0),
                )
                .unwrap(),
            },
        };

        let query_processor = VectorOperator::boxed(time_shift)
            .initialize(WorkflowOperatorPath::initialize_root(), &execution_context)
            .await
            .unwrap()
            .query_processor()
            .unwrap()
            .multi_point()
            .unwrap();

        let mut stream = query_processor
            .vector_query(
                VectorQueryRectangle::new(
                    BoundingBox2D::new((0., 0.).into(), (2., 2.).into()).unwrap(),
                    TimeInterval::new(
                        DateTime::new_utc(2009, 1, 1, 0, 0, 0),
                        DateTime::new_utc(2012, 1, 1, 0, 0, 0),
                    )
                    .unwrap(),
                    ColumnSelection::all(),
                ),
                &query_context,
            )
            .await
            .unwrap();

        let mut result = Vec::new();
        while let Some(collection) = stream.next().await {
            result.push(collection.unwrap());
        }

        assert_eq!(result.len(), 1);

        let expected = MultiPointCollection::from_data(
            MultiPoint::many(vec![(0., 0.)]).unwrap(),
            vec![
                TimeInterval::new(
                    DateTime::new_utc(2009, 1, 1, 0, 0, 0),
                    DateTime::new_utc_with_millis(2013, 8, 1, 23, 59, 59, 999),
                )
                .unwrap(),
            ],
            Default::default(),
            CacheHint::default(),
        )
        .unwrap();

        assert!(result[0].chunks_equal_ignoring_cache_hint(&expected));
    }

    #[tokio::test]
    async fn test_relative_vector_shift() {
        let execution_context = MockExecutionContext::test_default();
        let query_context = execution_context.mock_query_context_test_default();

        let source = MockFeatureCollectionSource::single(
            MultiPointCollection::from_data(
                MultiPoint::many(vec![(0., 0.), (1., 1.), (2., 2.)]).unwrap(),
                vec![
                    TimeInterval::new(
                        DateTime::new_utc(2009, 1, 1, 0, 0, 0),
                        DateTime::new_utc_with_millis(2010, 12, 31, 23, 59, 59, 999),
                    )
                    .unwrap(),
                    TimeInterval::new(
                        DateTime::new_utc(2009, 6, 3, 0, 0, 0),
                        DateTime::new_utc(2010, 7, 14, 0, 0, 0),
                    )
                    .unwrap(),
                    TimeInterval::new(
                        DateTime::new_utc(2010, 1, 1, 0, 0, 0),
                        DateTime::new_utc_with_millis(2011, 3, 31, 23, 59, 59, 999),
                    )
                    .unwrap(),
                ],
                Default::default(),
                CacheHint::default(),
            )
            .unwrap(),
        );

        let time_shift = TimeShift {
            sources: SingleRasterOrVectorSource {
                source: RasterOrVectorOperator::Vector(source.boxed()),
            },
            params: TimeShiftParams::Relative {
                granularity: TimeGranularity::Years,
                value: -1,
            },
        };

        let query_processor = VectorOperator::boxed(time_shift)
            .initialize(WorkflowOperatorPath::initialize_root(), &execution_context)
            .await
            .unwrap()
            .query_processor()
            .unwrap()
            .multi_point()
            .unwrap();

        let mut stream = query_processor
            .vector_query(
                VectorQueryRectangle::new(
                    BoundingBox2D::new((0., 0.).into(), (2., 2.).into()).unwrap(),
                    TimeInterval::new(
                        DateTime::new_utc(2010, 1, 1, 0, 0, 0),
                        DateTime::new_utc(2011, 1, 1, 0, 0, 0),
                    )
                    .unwrap(),
                    ColumnSelection::all(),
                ),
                &query_context,
            )
            .await
            .unwrap();

        let mut result = Vec::new();
        while let Some(collection) = stream.next().await {
            result.push(collection.unwrap());
        }

        assert_eq!(result.len(), 1);

        let expected = MultiPointCollection::from_data(
            MultiPoint::many(vec![(0., 0.), (1., 1.)]).unwrap(),
            vec![
                TimeInterval::new(
                    DateTime::new_utc(2010, 1, 1, 0, 0, 0),
                    DateTime::new_utc_with_millis(2011, 12, 31, 23, 59, 59, 999),
                )
                .unwrap(),
                TimeInterval::new(
                    DateTime::new_utc(2010, 6, 3, 0, 0, 0),
                    DateTime::new_utc(2011, 7, 14, 0, 0, 0),
                )
                .unwrap(),
            ],
            Default::default(),
            CacheHint::default(),
        )
        .unwrap();

        assert!(result[0].chunks_equal_ignoring_cache_hint(&expected));
    }

    #[tokio::test]
    #[allow(clippy::too_many_lines)]
    async fn test_absolute_raster_shift() {
        let tile_size_in_pixels = GridShape2D::new_2d(3, 2);
        let result_descriptor = RasterResultDescriptor {
            data_type: RasterDataType::U8,
            spatial_reference: SpatialReference::epsg_4326().into(),
            time: TimeDescriptor::new_regular_with_epoch(
                Some(TimeInterval::new_unchecked(
                    DateTime::new_utc(2010, 1, 1, 0, 0, 0),
                    DateTime::new_utc(2013, 1, 1, 0, 0, 0),
                )),
                TimeStep::years(1).unwrap(),
            ),
            spatial_grid: SpatialGridDescriptor::source_from_parts(
                GeoTransform::new(Coordinate2D::new(0., -3.), 1., -1.),
                GridShape2D::new_2d(3, 4).bounding_box(),
            ),
            bands: RasterBandDescriptors::new_single_band(),
        };
        let tiling_specification = TilingSpecification::new(tile_size_in_pixels);

        let empty_grid = GridOrEmpty::Empty(EmptyGrid2D::<u8>::new(tile_size_in_pixels));
        let raster_tiles = vec![
            RasterTile2D::new_with_tile_info(
                TimeInterval::new_unchecked(
                    DateTime::new_utc(2010, 1, 1, 0, 0, 0),
                    DateTime::new_utc(2011, 1, 1, 0, 0, 0),
                ),
                TileInformation {
                    global_tile_position: [-1, 0].into(),
                    tile_size_in_pixels: [3, 2].into(),
                    global_geo_transform: TestDefault::test_default(),
                },
                0,
                empty_grid.clone(),
                CacheHint::default(),
            ),
            RasterTile2D::new_with_tile_info(
                TimeInterval::new_unchecked(
                    DateTime::new_utc(2010, 1, 1, 0, 0, 0),
                    DateTime::new_utc(2011, 1, 1, 0, 0, 0),
                ),
                TileInformation {
                    global_tile_position: [-1, 1].into(),
                    tile_size_in_pixels: [3, 2].into(),
                    global_geo_transform: TestDefault::test_default(),
                },
                0,
                empty_grid.clone(),
                CacheHint::default(),
            ),
            RasterTile2D::new_with_tile_info(
                TimeInterval::new_unchecked(
                    DateTime::new_utc(2011, 1, 1, 0, 0, 0),
                    DateTime::new_utc(2012, 1, 1, 0, 0, 0),
                ),
                TileInformation {
                    global_tile_position: [-1, 0].into(),
                    tile_size_in_pixels: [3, 2].into(),
                    global_geo_transform: TestDefault::test_default(),
                },
                0,
                empty_grid.clone(),
                CacheHint::default(),
            ),
            RasterTile2D::new_with_tile_info(
                TimeInterval::new_unchecked(
                    DateTime::new_utc(2011, 1, 1, 0, 0, 0),
                    DateTime::new_utc(2012, 1, 1, 0, 0, 0),
                ),
                TileInformation {
                    global_tile_position: [-1, 1].into(),
                    tile_size_in_pixels: [3, 2].into(),
                    global_geo_transform: TestDefault::test_default(),
                },
                0,
                empty_grid.clone(),
                CacheHint::default(),
            ),
            RasterTile2D::new_with_tile_info(
                TimeInterval::new_unchecked(
                    DateTime::new_utc(2012, 1, 1, 0, 0, 0),
                    DateTime::new_utc(2013, 1, 1, 0, 0, 0),
                ),
                TileInformation {
                    global_tile_position: [-1, 0].into(),
                    tile_size_in_pixels: [3, 2].into(),
                    global_geo_transform: TestDefault::test_default(),
                },
                0,
                empty_grid.clone(),
                CacheHint::default(),
            ),
            RasterTile2D::new_with_tile_info(
                TimeInterval::new_unchecked(
                    DateTime::new_utc(2012, 1, 1, 0, 0, 0),
                    DateTime::new_utc(2013, 1, 1, 0, 0, 0),
                ),
                TileInformation {
                    global_tile_position: [-1, 1].into(),
                    tile_size_in_pixels: [3, 2].into(),
                    global_geo_transform: TestDefault::test_default(),
                },
                0,
                empty_grid.clone(),
                CacheHint::default(),
            ),
        ];

        let mrs = MockRasterSource {
            params: MockRasterSourceParams {
                data: raster_tiles,
                result_descriptor,
            },
        }
        .boxed();

        let time_shift = TimeShift {
            sources: SingleRasterOrVectorSource {
                source: RasterOrVectorOperator::Raster(mrs),
            },
            params: TimeShiftParams::Absolute {
                time_interval: TimeInterval::new_unchecked(
                    DateTime::new_utc(2011, 1, 1, 0, 0, 0),
                    DateTime::new_utc(2012, 1, 1, 0, 0, 0),
                ),
            },
        };

        let execution_context = MockExecutionContext::new_with_tiling_spec(tiling_specification);

        let query_context = execution_context.mock_query_context_test_default();

        let query_processor = RasterOperator::boxed(time_shift)
            .initialize(WorkflowOperatorPath::initialize_root(), &execution_context)
            .await
            .unwrap()
            .query_processor()
            .unwrap()
            .get_u8()
            .unwrap();

        let mut stream = query_processor
            .raster_query(
                RasterQueryRectangle::new(
                    GridBoundingBox2D::new([-3, 0], [-1, 3]).unwrap(),
                    TimeInterval::new(
                        DateTime::new_utc(2010, 1, 1, 0, 0, 0),
                        DateTime::new_utc(2011, 1, 1, 0, 0, 0),
                    )
                    .unwrap(),
                    BandSelection::first(),
                ),
                &query_context,
            )
            .await
            .unwrap();

        let mut result = Vec::new();
        while let Some(tile) = stream.next().await {
            result.push(tile.unwrap());
        }

        assert_eq!(result.len(), 2);

        assert_eq!(
            result[0].time,
            TimeInterval::new_unchecked(
                DateTime::new_utc(2010, 1, 1, 0, 0, 0),
                DateTime::new_utc(2011, 1, 1, 0, 0, 0),
            ),
        );
        assert_eq!(
            result[1].time,
            TimeInterval::new_unchecked(
                DateTime::new_utc(2010, 1, 1, 0, 0, 0),
                DateTime::new_utc(2011, 1, 1, 0, 0, 0),
            ),
        );
    }

    #[tokio::test]
    #[allow(clippy::too_many_lines)]
    async fn test_relative_raster_shift() {
        let tile_size_in_pixels = GridShape2D::new_2d(3, 2);
        let result_descriptor = RasterResultDescriptor {
            data_type: RasterDataType::U8,
            spatial_reference: SpatialReference::epsg_4326().into(),
            time: TimeDescriptor::new_regular_with_epoch(
                Some(TimeInterval::new_unchecked(
                    DateTime::new_utc(2010, 1, 1, 0, 0, 0),
                    DateTime::new_utc(2013, 1, 1, 0, 0, 0),
                )),
                TimeStep::years(1).unwrap(),
            ),
            spatial_grid: SpatialGridDescriptor::source_from_parts(
                GeoTransform::new(Coordinate2D::new(0., 0.), 1., -1.),
                GridBoundingBox2D::new([-3, 0], [0, 4]).unwrap(),
            ),
            bands: RasterBandDescriptors::new_single_band(),
        };
        let tiling_specification = TilingSpecification::new(tile_size_in_pixels);

        let empty_grid = GridOrEmpty::Empty(EmptyGrid2D::<u8>::new(tile_size_in_pixels));
        let raster_tiles = vec![
            RasterTile2D::new_with_tile_info(
                TimeInterval::new_unchecked(
                    DateTime::new_utc(2010, 1, 1, 0, 0, 0),
                    DateTime::new_utc(2011, 1, 1, 0, 0, 0),
                ),
                TileInformation {
                    global_tile_position: [-1, 0].into(),
                    tile_size_in_pixels: [3, 2].into(),
                    global_geo_transform: TestDefault::test_default(),
                },
                0,
                empty_grid.clone(),
                CacheHint::default(),
            ),
            RasterTile2D::new_with_tile_info(
                TimeInterval::new_unchecked(
                    DateTime::new_utc(2010, 1, 1, 0, 0, 0),
                    DateTime::new_utc(2011, 1, 1, 0, 0, 0),
                ),
                TileInformation {
                    global_tile_position: [-1, 1].into(),
                    tile_size_in_pixels: [3, 2].into(),
                    global_geo_transform: TestDefault::test_default(),
                },
                0,
                empty_grid.clone(),
                CacheHint::default(),
            ),
            RasterTile2D::new_with_tile_info(
                TimeInterval::new_unchecked(
                    DateTime::new_utc(2011, 1, 1, 0, 0, 0),
                    DateTime::new_utc(2012, 1, 1, 0, 0, 0),
                ),
                TileInformation {
                    global_tile_position: [-1, 0].into(),
                    tile_size_in_pixels: [3, 2].into(),
                    global_geo_transform: TestDefault::test_default(),
                },
                0,
                empty_grid.clone(),
                CacheHint::default(),
            ),
            RasterTile2D::new_with_tile_info(
                TimeInterval::new_unchecked(
                    DateTime::new_utc(2011, 1, 1, 0, 0, 0),
                    DateTime::new_utc(2012, 1, 1, 0, 0, 0),
                ),
                TileInformation {
                    global_tile_position: [-1, 1].into(),
                    tile_size_in_pixels: [3, 2].into(),
                    global_geo_transform: TestDefault::test_default(),
                },
                0,
                empty_grid.clone(),
                CacheHint::default(),
            ),
            RasterTile2D::new_with_tile_info(
                TimeInterval::new_unchecked(
                    DateTime::new_utc(2012, 1, 1, 0, 0, 0),
                    DateTime::new_utc(2013, 1, 1, 0, 0, 0),
                ),
                TileInformation {
                    global_tile_position: [-1, 0].into(),
                    tile_size_in_pixels: [3, 2].into(),
                    global_geo_transform: TestDefault::test_default(),
                },
                0,
                empty_grid.clone(),
                CacheHint::default(),
            ),
            RasterTile2D::new_with_tile_info(
                TimeInterval::new_unchecked(
                    DateTime::new_utc(2012, 1, 1, 0, 0, 0),
                    DateTime::new_utc(2013, 1, 1, 0, 0, 0),
                ),
                TileInformation {
                    global_tile_position: [-1, 1].into(),
                    tile_size_in_pixels: [3, 2].into(),
                    global_geo_transform: TestDefault::test_default(),
                },
                0,
                empty_grid.clone(),
                CacheHint::default(),
            ),
        ];

        let mrs = MockRasterSource {
            params: MockRasterSourceParams {
                data: raster_tiles,
                result_descriptor,
            },
        }
        .boxed();

        let time_shift = TimeShift {
            sources: SingleRasterOrVectorSource {
                source: RasterOrVectorOperator::Raster(mrs),
            },
            params: TimeShiftParams::Relative {
                granularity: TimeGranularity::Years,
                value: 1,
            },
        };

        let execution_context = MockExecutionContext::new_with_tiling_spec(tiling_specification);
        let query_context = execution_context.mock_query_context_test_default();

        let query_processor = RasterOperator::boxed(time_shift)
            .initialize(WorkflowOperatorPath::initialize_root(), &execution_context)
            .await
            .unwrap()
            .query_processor()
            .unwrap()
            .get_u8()
            .unwrap();

        let mut stream = query_processor
            .raster_query(
                RasterQueryRectangle::new(
                    GridBoundingBox2D::new([-3, 0], [-1, 3]).unwrap(),
                    TimeInterval::new(
                        DateTime::new_utc(2010, 1, 1, 0, 0, 0),
                        DateTime::new_utc(2011, 1, 1, 0, 0, 0),
                    )
                    .unwrap(),
                    BandSelection::first(),
                ),
                &query_context,
            )
            .await
            .unwrap();

        let mut result = Vec::new();
        while let Some(tile) = stream.next().await {
            result.push(tile.unwrap());
        }

        assert_eq!(result.len(), 2);

        assert_eq!(
            result[0].time,
            TimeInterval::new_unchecked(
                DateTime::new_utc(2010, 1, 1, 0, 0, 0),
                DateTime::new_utc(2011, 1, 1, 0, 0, 0),
            ),
        );
        assert_eq!(
            result[1].time,
            TimeInterval::new_unchecked(
                DateTime::new_utc(2010, 1, 1, 0, 0, 0),
                DateTime::new_utc(2011, 1, 1, 0, 0, 0),
            ),
        );
    }

    #[tokio::test]
    async fn test_expression_on_shifted_raster() {
        let mut execution_context = MockExecutionContext::test_default();

        let ndvi_source = GdalSource {
            params: GdalSourceParameters::new(add_ndvi_dataset(&mut execution_context)),
        }
        .boxed();

        let shifted_ndvi_source = RasterOperator::boxed(TimeShift {
            params: TimeShiftParams::Relative {
                granularity: TimeGranularity::Months,
                value: -1,
            },
            sources: SingleRasterOrVectorSource {
                source: RasterOrVectorOperator::Raster(ndvi_source.clone()),
            },
        });

        let expression = Expression {
            params: ExpressionParams {
                expression: "A - B".to_string(),
                output_type: RasterDataType::F64,
                output_band: None,
                map_no_data: false,
            },
            sources: SingleRasterSource {
                raster: RasterStacker {
                    params: RasterStackerParams {
                        rename_bands: RenameBands::Default,
                    },
                    sources: MultipleRasterSources {
                        rasters: vec![ndvi_source, shifted_ndvi_source],
                    },
                }
                .boxed(),
            },
        }
        .boxed();

        let query_processor = expression
            .initialize(WorkflowOperatorPath::initialize_root(), &execution_context)
            .await
            .unwrap()
            .query_processor()
            .unwrap()
            .get_f64()
            .unwrap();

        let query_context = execution_context.mock_query_context_test_default();

        let mut stream = query_processor
            .raster_query(
                RasterQueryRectangle::new(
                    GridBoundingBox2D::new_min_max(-90, 89, -180, 179).unwrap(), // Note: this is not the actual bounding box of the NDVI dataset. The pixel size is 0.1!
                    TimeInterval::new_instant(DateTime::new_utc(2014, 3, 1, 0, 0, 0)).unwrap(),
                    BandSelection::first(),
                ),
                &query_context,
            )
            .await
            .unwrap();

        let mut result = Vec::new();
        while let Some(tile) = stream.next().await {
            result.push(tile.unwrap());
        }

        assert_eq!(result.len(), 4);
        assert_eq!(
            result[0].time,
            TimeInterval::new(
                DateTime::new_utc(2014, 3, 1, 0, 0, 0),
                DateTime::new_utc(2014, 4, 1, 0, 0, 0)
            )
            .unwrap()
        );
    }

    #[tokio::test]
    async fn test_expression_on_absolute_shifted_raster() {
        let mut execution_context = MockExecutionContext::test_default();

        let ndvi_source = GdalSource {
            params: GdalSourceParameters::new(add_ndvi_dataset(&mut execution_context)),
        }
        .boxed();

        let shifted_ndvi_source = RasterOperator::boxed(TimeShift {
            params: TimeShiftParams::Absolute {
                time_interval: TimeInterval::new_instant(DateTime::new_utc(2014, 5, 1, 0, 0, 0))
                    .unwrap(),
            },
            sources: SingleRasterOrVectorSource {
                source: RasterOrVectorOperator::Raster(ndvi_source),
            },
        });

        let query_processor = shifted_ndvi_source
            .initialize(WorkflowOperatorPath::initialize_root(), &execution_context)
            .await
            .unwrap()
            .query_processor()
            .unwrap()
            .get_u8()
            .unwrap();

        let query_context = execution_context.mock_query_context_test_default();

        let mut stream = query_processor
            .raster_query(
                RasterQueryRectangle::new(
                    GridBoundingBox2D::new_min_max(-90, 89, -180, 179).unwrap(), // Note: this is not the actual bounding box of the NDVI dataset. The pixel size is 0.1!
                    TimeInterval::new_instant(DateTime::new_utc(2014, 3, 1, 0, 0, 0)).unwrap(),
                    BandSelection::first(),
                ),
                &query_context,
            )
            .await
            .unwrap();

        let mut result = Vec::new();
        while let Some(tile) = stream.next().await {
            result.push(tile.unwrap());
        }

        assert_eq!(result.len(), 4);
        assert_eq!(
            result[0].time,
            TimeInterval::new(
                DateTime::new_utc(2014, 3, 1, 0, 0, 0),
                DateTime::new_utc(2014, 4, 1, 0, 0, 0)
            )
            .unwrap()
        );
    }

    #[test]
    fn shift_eternal() {
        let eternal = TimeInterval::default();

        let f_shift = RelativeForwardShift {
            step: TimeStep {
                granularity: TimeGranularity::Seconds,
                step: 1,
            },
        };

        let (shifted, state) = f_shift.shift(eternal).unwrap();
        assert_eq!(shifted, eternal);
        let reverse_shifted = f_shift.reverse_shift(eternal, state).unwrap();
        assert_eq!(reverse_shifted, eternal);

        let b_shift = RelativeBackwardShift {
            step: TimeStep {
                granularity: TimeGranularity::Seconds,
                step: 1,
            },
        };

        let (shifted, state) = b_shift.shift(eternal).unwrap();
        assert_eq!(shifted, eternal);

        let reverse_shifted = b_shift.reverse_shift(eternal, state).unwrap();
        assert_eq!(reverse_shifted, eternal);
    }

    #[test]
    fn shift_begin_of_time() {
        let time = TimeInterval::new_unchecked(TimeInstance::MIN, TimeInstance::EPOCH_START);

        let f_shift = RelativeForwardShift {
            step: TimeStep {
                granularity: TimeGranularity::Seconds,
                step: 1,
            },
        };

        let expected_shift =
            TimeInterval::new_unchecked(TimeInstance::MIN, TimeInstance::EPOCH_START + 1_000);

        let (shifted, state) = f_shift.shift(time).unwrap();
        assert_eq!(shifted, expected_shift);

        let reverse_shifted = f_shift.reverse_shift(expected_shift, state).unwrap();
        assert_eq!(reverse_shifted, time);

        let f_shift = RelativeBackwardShift {
            step: TimeStep {
                granularity: TimeGranularity::Seconds,
                step: 1,
            },
        };

        let expected_shift =
            TimeInterval::new_unchecked(TimeInstance::MIN, TimeInstance::EPOCH_START - 1_000);

        let (shifted, state) = f_shift.shift(time).unwrap();
        assert_eq!(shifted, expected_shift);

        let reverse_shifted = f_shift.reverse_shift(expected_shift, state).unwrap();
        assert_eq!(reverse_shifted, time);
    }

    #[test]
    fn shift_end_of_time() {
        let time = TimeInterval::new_unchecked(TimeInstance::EPOCH_START, TimeInstance::MAX);

        let f_shift = RelativeForwardShift {
            step: TimeStep {
                granularity: TimeGranularity::Seconds,
                step: 1,
            },
        };

        let expected_shift =
            TimeInterval::new_unchecked(TimeInstance::EPOCH_START + 1_000, TimeInstance::MAX);

        let (shifted, state) = f_shift.shift(time).unwrap();
        assert_eq!(shifted, expected_shift);

        let reverse_shifted = f_shift.reverse_shift(expected_shift, state).unwrap();
        assert_eq!(reverse_shifted, time);

        let f_shift = RelativeBackwardShift {
            step: TimeStep {
                granularity: TimeGranularity::Seconds,
                step: 1,
            },
        };

        let expected_shift =
            TimeInterval::new_unchecked(TimeInstance::EPOCH_START - 1_000, TimeInstance::MAX);

        let (shifted, state) = f_shift.shift(time).unwrap();
        assert_eq!(shifted, expected_shift);

        let reverse_shifted = f_shift.reverse_shift(expected_shift, state).unwrap();
        assert_eq!(reverse_shifted, time);
    }
}

geo-engine / geoengine / 19438548963

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