17044350441

Committed 18 Aug 2025 03:00PM UTC coverage: 88.074%. First build

Build # 17044350441

Build Type

Pull #1015

github

Committed by

web-flow

Commit Message

Merge f38e4bbee into db8685e5e

Pull Request Pull Request #1015: workflow optimization (wip)

Coverage Stats

1666 of 2178 new or added lines in 51 files covered. (76.49%)

114929 of 130492 relevant lines covered (88.07%)

179856.12 hits per line

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

93.65

/operators/src/source/csv.rs

use crate::engine::{
    CanonicOperatorName, InitializedVectorOperator, OperatorData, OperatorName, QueryContext,
    SourceOperator, TypedVectorQueryProcessor, VectorOperator, VectorQueryProcessor,
    VectorResultDescriptor,
};
use crate::engine::{QueryProcessor, WorkflowOperatorPath};
use crate::error;
use crate::optimization::OptimizationError;
use crate::util::{Result, safe_lock_mutex};
use async_trait::async_trait;
use csv::{Position, Reader, StringRecord};
use futures::stream::BoxStream;
use futures::task::{Context, Poll};
use futures::{Stream, StreamExt};
use geoengine_datatypes::collections::{
    BuilderProvider, GeoFeatureCollectionRowBuilder, MultiPointCollection, VectorDataType,
};
use geoengine_datatypes::dataset::NamedData;
use geoengine_datatypes::primitives::SpatialResolution;
use geoengine_datatypes::{
    primitives::{
        BoundingBox2D, ColumnSelection, Coordinate2D, TimeInterval, VectorQueryRectangle,
    },
    spatial_reference::SpatialReference,
};
use serde::{Deserialize, Serialize};
use snafu::{OptionExt, ResultExt, ensure};
use std::path::PathBuf;
use std::pin::Pin;
use std::sync::atomic::Ordering;
use std::sync::{Arc, Mutex};
use std::{fs::File, sync::atomic::AtomicBool};

/// Parameters for the CSV Source Operator
#[derive(Debug, PartialEq, Eq, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
pub struct CsvSourceParameters {
    pub file_path: PathBuf,
    pub field_separator: char,
    pub geometry: CsvGeometrySpecification,
    #[serde(default)]
    pub time: CsvTimeSpecification,
}

#[derive(Debug, PartialEq, Eq, Serialize, Deserialize, Clone)]
#[serde(tag = "type", rename_all = "lowercase")]
pub enum CsvGeometrySpecification {
    #[allow(clippy::upper_case_acronyms)]
    XY { x: String, y: String },
}

#[derive(Debug, PartialEq, Eq, Serialize, Deserialize, Clone)]
pub enum CsvTimeSpecification {
    None,
}

impl Default for CsvTimeSpecification {
    fn default() -> Self {
        Self::None
    }
}

enum ReaderState {
    Untouched(Reader<File>),
    OnGoing {
        header: ParsedHeader,
        records: csv::StringRecordsIntoIter<File>,
    },
    Error,
}

impl ReaderState {
    pub fn setup_once(&mut self, geometry_specification: CsvGeometrySpecification) -> Result<()> {
        if let ReaderState::Untouched(..) = self {
            // pass
        } else {
            return Ok(());
        }

        let old_state = std::mem::replace(self, ReaderState::Error);

        if let ReaderState::Untouched(mut csv_reader) = old_state {
            let header = CsvSourceStream::setup_read(geometry_specification, &mut csv_reader)?;

            let mut records = csv_reader.into_records();

            // consume the first row, which is the header
            if header.has_header {
                // TODO: throw error
                records.next();
            }

            *self = ReaderState::OnGoing { header, records }
        }

        Ok(())
    }
}

pub struct CsvSourceStream {
    parameters: CsvSourceParameters,
    bbox: BoundingBox2D,
    chunk_size: usize,
    reader_state: Arc<Mutex<ReaderState>>,
    thread_is_computing: Arc<AtomicBool>,
    #[allow(clippy::option_option)]
    poll_result: Arc<Mutex<Option<Option<Result<MultiPointCollection>>>>>,
}

pub type CsvSource = SourceOperator<CsvSourceParameters>;

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

impl OperatorData for CsvSourceParameters {
    fn data_names_collect(&self, _data_names: &mut Vec<NamedData>) {}
}

#[typetag::serde]
#[async_trait]
impl VectorOperator for CsvSource {
    async fn _initialize(
        self: Box<Self>,
        path: WorkflowOperatorPath,
        _context: &dyn crate::engine::ExecutionContext,
    ) -> Result<Box<dyn InitializedVectorOperator>> {
        let initialized_source = InitializedCsvSource {
            name: CanonicOperatorName::from(&self),
            path,
            result_descriptor: VectorResultDescriptor {
                data_type: VectorDataType::MultiPoint, // TODO: get as user input
                spatial_reference: SpatialReference::epsg_4326().into(), // TODO: get as user input
                columns: Default::default(), // TODO: get when source allows loading other columns
                time: None,
                bbox: None,
            },
            state: self.params,
        };

        Ok(initialized_source.boxed())
    }

    span_fn!(CsvSource);
}

pub struct InitializedCsvSource {
    name: CanonicOperatorName,
    path: WorkflowOperatorPath,
    result_descriptor: VectorResultDescriptor,
    state: CsvSourceParameters,
}

impl InitializedVectorOperator for InitializedCsvSource {
    fn query_processor(&self) -> Result<crate::engine::TypedVectorQueryProcessor> {
        Ok(TypedVectorQueryProcessor::MultiPoint(
            CsvSourceProcessor {
                params: self.state.clone(),
                result_descriptor: self.result_descriptor.clone(),
            }
            .boxed(),
        ))
    }

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

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

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

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

    fn optimize(
        &self,
        _target_resolution: SpatialResolution,
    ) -> Result<Box<dyn VectorOperator>, OptimizationError> {
        Ok(CsvSource {
            params: self.state.clone(),
        }
        .boxed())
    }
}

impl CsvSourceStream {
    /// Creates a new `CsvSource`
    ///
    /// # Errors
    ///
    /// This constructor fails if the delimiter is not an ASCII character.
    /// Furthermore, there are IO errors from the reader.
    ///
    // TODO: include time interval, e.g. QueryRectangle parameter
    pub fn new(
        parameters: CsvSourceParameters,
        bbox: BoundingBox2D,
        chunk_size: usize,
    ) -> Result<Self> {
        ensure!(
            parameters.field_separator.is_ascii(),
            error::CsvSource {
                details: "Delimiter must be ASCII character"
            }
        );

        Ok(Self {
            reader_state: Arc::new(Mutex::new(ReaderState::Untouched(
                csv::ReaderBuilder::new()
                    .delimiter(parameters.field_separator as u8)
                    .has_headers(true)
                    .from_path(parameters.file_path.as_path())
                    .context(error::CsvSourceReader {})?,
            ))),
            thread_is_computing: Arc::new(AtomicBool::new(false)),
            poll_result: Arc::new(Mutex::new(None)),
            parameters,
            bbox,
            chunk_size,
        })
    }

    fn setup_read(
        geometry_specification: CsvGeometrySpecification,
        csv_reader: &mut Reader<File>,
    ) -> Result<ParsedHeader> {
        csv_reader
            .seek(Position::new())
            .context(error::CsvSourceReader {})?; // start at beginning

        ensure!(
            csv_reader.has_headers(),
            error::CsvSource {
                details: "CSV file must contain header",
            }
        );

        let header = csv_reader.headers().context(error::CsvSourceReader)?;

        let CsvGeometrySpecification::XY { x, y } = geometry_specification;
        let x_index = header
            .iter()
            .position(|v| v == x)
            .context(error::CsvSource {
                details: "Cannot find x index in csv header",
            })?;
        let y_index = header
            .iter()
            .position(|v| v == y)
            .context(error::CsvSource {
                details: "Cannot find y index in csv header",
            })?;

        Ok(ParsedHeader {
            has_header: true,
            x_index,
            y_index,
        })
    }

    /// Parse a single CSV row
    fn parse_row(header: &ParsedHeader, row: &StringRecord) -> Result<ParsedRow> {
        let x: f64 = row
            .get(header.x_index)
            .context(error::CsvSource {
                details: "Cannot find x index key",
            })?
            .parse()
            .map_err(|_error| error::Error::CsvSource {
                details: "Cannot parse x coordinate".to_string(),
            })?;
        let y: f64 = row
            .get(header.y_index)
            .context(error::CsvSource {
                details: "Cannot find y index key",
            })?
            .parse()
            .map_err(|_error| error::Error::CsvSource {
                details: "Cannot parse y coordinate".to_string(),
            })?;

        Ok(ParsedRow {
            coordinate: (x, y).into(),
            time_interval: TimeInterval::default(),
        })
    }
}

impl Stream for CsvSourceStream {
    type Item = Result<MultiPointCollection>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        // TODO: handle lock poisoning on multiple occasions

        if self.thread_is_computing.load(Ordering::Relaxed) {
            return Poll::Pending;
        }

        let mut poll_result = safe_lock_mutex(&self.poll_result);
        if let Some(x) = poll_result.take() {
            return Poll::Ready(x);
        }

        self.thread_is_computing.store(true, Ordering::Relaxed);

        let is_working = self.thread_is_computing.clone();
        let reader_state = self.reader_state.clone();
        let poll_result = self.poll_result.clone();

        let bbox = self.bbox;
        let chunk_size = self.chunk_size;
        let parameters = self.parameters.clone();
        let waker = cx.waker().clone();

        crate::util::spawn_blocking(move || {
            let mut csv_reader = safe_lock_mutex(&reader_state);
            let computation_result = || -> Result<Option<MultiPointCollection>> {
                // TODO: is clone necessary?
                let geometry_specification = parameters.geometry.clone();
                csv_reader.setup_once(geometry_specification)?;

                let (header, records) = match &mut *csv_reader {
                    ReaderState::OnGoing { header, records } => (header, records),
                    ReaderState::Error => return Ok(None),
                    ReaderState::Untouched(_) => unreachable!(),
                };

                let mut builder = MultiPointCollection::builder().finish_header();
                let mut number_of_entries = 0; // TODO: add size/len to builder

                while number_of_entries < chunk_size {
                    let Some(record) = records.next() else {
                        break;
                    };

                    let row = record.context(error::CsvSourceReader)?;
                    let parsed_row = CsvSourceStream::parse_row(header, &row)?;

                    // TODO: filter time
                    if bbox.contains_coordinate(&parsed_row.coordinate) {
                        builder.push_geometry(parsed_row.coordinate.into());
                        builder.push_time_interval(parsed_row.time_interval);
                        builder.finish_row();

                        number_of_entries += 1;
                    }
                }

                // TODO: is this the correct cancellation criterion?
                if number_of_entries > 0 {
                    let collection = builder.build()?;
                    Ok(Some(collection))
                } else {
                    Ok(None)
                }
            }();

            *safe_lock_mutex(&poll_result) = Some(match computation_result {
                Ok(Some(collection)) => Some(Ok(collection)),
                Ok(None) => None,
                Err(e) => Some(Err(e)),
            });
            is_working.store(false, Ordering::Relaxed);

            waker.wake();
        });

        Poll::Pending
    }
}

#[derive(Debug)]
struct CsvSourceProcessor {
    params: CsvSourceParameters,
    result_descriptor: VectorResultDescriptor,
}

#[async_trait]
impl QueryProcessor for CsvSourceProcessor {
    type Output = MultiPointCollection;
    type SpatialBounds = BoundingBox2D;
    type Selection = ColumnSelection;
    type ResultDescription = VectorResultDescriptor;

    async fn _query<'a>(
        &'a self,
        query: VectorQueryRectangle,
        _ctx: &'a dyn QueryContext,
    ) -> Result<BoxStream<'a, Result<Self::Output>>> {
        // TODO: properly handle chunk_size
        Ok(CsvSourceStream::new(self.params.clone(), query.spatial_bounds(), 10)?.boxed())
    }

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

#[derive(Clone, Copy, Debug)]
struct ParsedHeader {
    pub has_header: bool,
    pub x_index: usize,
    pub y_index: usize,
}

struct ParsedRow {
    pub coordinate: Coordinate2D,
    pub time_interval: TimeInterval,
    // TODO: fields
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::engine::MockExecutionContext;
    use geoengine_datatypes::{
        collections::{FeatureCollectionInfos, ToGeoJson},
        util::test::TestDefault,
    };
    use std::io::{Seek, SeekFrom, Write};

    #[test]
    fn it_deserializes() {
        let json_string = r#"
            {
                "type": "CsvSource",
                "params": {
                    "filePath": "/foo/bar.csv",
                    "fieldSeparator": ",",
                    "geometry": {
                        "type": "xy",
                        "x": "x",
                        "y": "y"
                    }
                }
            }"#;

        let operator: CsvSource = serde_json::from_str(json_string).unwrap();

        assert_eq!(
            operator,
            CsvSource {
                params: CsvSourceParameters {
                    file_path: "/foo/bar.csv".into(),
                    field_separator: ',',
                    geometry: CsvGeometrySpecification::XY {
                        x: "x".into(),
                        y: "y".into()
                    },
                    time: CsvTimeSpecification::None,
                },
            }
        );
    }

    #[tokio::test]
    async fn read_points() {
        let mut fake_file = tempfile::NamedTempFile::new().unwrap();
        write!(
            fake_file,
            "\
x,y
0,1
2,3
4,5
"
        )
        .unwrap();
        fake_file.seek(SeekFrom::Start(0)).unwrap();

        let mut csv_source = CsvSourceStream::new(
            CsvSourceParameters {
                file_path: fake_file.path().into(),
                field_separator: ',',
                geometry: CsvGeometrySpecification::XY {
                    x: "x".into(),
                    y: "y".into(),
                },
                time: CsvTimeSpecification::None,
            },
            BoundingBox2D::new_unchecked((0., 0.).into(), (5., 5.).into()),
            2,
        )
        .unwrap();

        assert_eq!(csv_source.next().await.unwrap().unwrap().len(), 2);
        assert_eq!(csv_source.next().await.unwrap().unwrap().len(), 1);
        assert!(csv_source.next().await.is_none());
    }

    #[tokio::test]
    async fn erroneous_point_rows() {
        let mut fake_file = tempfile::NamedTempFile::new().unwrap();
        write!(
            fake_file,
            "\
x,y
0,1
CORRUPT
4,5
"
        )
        .unwrap();
        fake_file.seek(SeekFrom::Start(0)).unwrap();

        let mut csv_source = CsvSourceStream::new(
            CsvSourceParameters {
                file_path: fake_file.path().into(),
                field_separator: ',',
                geometry: CsvGeometrySpecification::XY {
                    x: "x".into(),
                    y: "y".into(),
                },
                time: CsvTimeSpecification::None,
            },
            BoundingBox2D::new_unchecked((0., 0.).into(), (5., 5.).into()),
            1,
        )
        .unwrap();

        assert_eq!(csv_source.next().await.unwrap().unwrap().len(), 1);
        assert!(csv_source.next().await.unwrap().is_err());
        assert_eq!(csv_source.next().await.unwrap().unwrap().len(), 1);
        assert!(csv_source.next().await.is_none());
    }

    #[tokio::test]
    async fn corrupt_point_header() {
        let mut fake_file = tempfile::NamedTempFile::new().unwrap();
        write!(
            fake_file,
            "\
x,z
0,1
2,3
4,5
"
        )
        .unwrap();
        fake_file.seek(SeekFrom::Start(0)).unwrap();

        let mut csv_source = CsvSourceStream::new(
            CsvSourceParameters {
                file_path: fake_file.path().into(),
                field_separator: ',',
                geometry: CsvGeometrySpecification::XY {
                    x: "x".into(),
                    y: "y".into(),
                },
                time: CsvTimeSpecification::None,
            },
            BoundingBox2D::new_unchecked((0., 0.).into(), (5., 5.).into()),
            1,
        )
        .unwrap();

        assert!(csv_source.next().await.unwrap().is_err());
        assert!(csv_source.next().await.is_none());
    }

    #[tokio::test]
    async fn processor() {
        let mut fake_file = tempfile::NamedTempFile::new().unwrap();
        write!(
            fake_file,
            "\
x,y
0,1
2,3
4,5
"
        )
        .unwrap();
        fake_file.seek(SeekFrom::Start(0)).unwrap();

        let params = CsvSourceParameters {
            file_path: fake_file.path().into(),
            field_separator: ',',
            geometry: CsvGeometrySpecification::XY {
                x: "x".into(),
                y: "y".into(),
            },
            time: CsvTimeSpecification::None,
        };

        let p = CsvSourceProcessor {
            params,
            result_descriptor: VectorResultDescriptor {
                data_type: VectorDataType::MultiPoint,
                spatial_reference: SpatialReference::epsg_4326().into(),
                columns: Default::default(),
                time: None,
                bbox: None,
            },
        };

        let query = VectorQueryRectangle::new(
            BoundingBox2D::new_unchecked(Coordinate2D::new(0., 0.), Coordinate2D::new(3., 3.)),
            TimeInterval::new_unchecked(0, 1),
            ColumnSelection::all(),
        );
        let ecx = MockExecutionContext::test_default();
        let ctx = ecx.mock_query_context((10 * 8 * 2).into());

        let r: Vec<Result<MultiPointCollection>> =
            p.query(query, &ctx).await.unwrap().collect().await;

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

        assert_eq!(
            serde_json::from_str::<serde_json::Value>(&r[0].as_ref().unwrap().to_geo_json())
                .unwrap(),
            serde_json::json!({
                "type": "FeatureCollection",
                "features": [{
                    "type": "Feature",
                    "geometry": {
                        "type": "Point",
                        "coordinates": [0.0, 1.0]
                    },
                    "properties": {},
                    "when": {
                        "start": "-262143-01-01T00:00:00+00:00",
                        "end": "+262142-12-31T23:59:59.999+00:00",
                        "type": "Interval"
                    }
                }, {
                    "type": "Feature",
                    "geometry": {
                        "type": "Point",
                        "coordinates": [2.0, 3.0]
                    },
                    "properties": {},
                    "when": {
                        "start": "-262143-01-01T00:00:00+00:00",
                        "end": "+262142-12-31T23:59:59.999+00:00",
                        "type": "Interval"
                    }
                }]
            })
        );
    }

    #[test]
    fn operator() {
        let mut temp_file = tempfile::NamedTempFile::new().unwrap();
        write!(
            temp_file,
            "\
x;y
0;1
2;3
4;5
"
        )
        .unwrap();
        temp_file.seek(SeekFrom::Start(0)).unwrap();

        let params = CsvSourceParameters {
            file_path: temp_file.path().into(),
            field_separator: ';',
            geometry: CsvGeometrySpecification::XY {
                x: "x".into(),
                y: "y".into(),
            },
            time: CsvTimeSpecification::None,
        };

        let operator = CsvSource { params }.boxed();

        let operator_json = serde_json::to_value(&operator).unwrap();

        assert_eq!(
            operator_json,
            serde_json::json!({
                "type": "CsvSource",
                "params": {
                    "filePath": temp_file.path(),
                    "fieldSeparator": ";",
                    "geometry": {
                        "type": "xy",
                        "x": "x",
                        "y": "y"
                    },
                    "time": "None"
                }
            })
        );

        let _operator: Box<dyn VectorOperator> = serde_json::from_value(operator_json).unwrap();
    }
}

1	use crate::engine::{
2	CanonicOperatorName, InitializedVectorOperator, OperatorData, OperatorName, QueryContext,
3	SourceOperator, TypedVectorQueryProcessor, VectorOperator, VectorQueryProcessor,
4	VectorResultDescriptor,
5	};
6	use crate::engine::{QueryProcessor, WorkflowOperatorPath};
7	use crate::error;
8	use crate::optimization::OptimizationError;
9	use crate::util::{Result, safe_lock_mutex};
10	use async_trait::async_trait;
11	use csv::{Position, Reader, StringRecord};
12	use futures::stream::BoxStream;
13	use futures::task::{Context, Poll};
14	use futures::{Stream, StreamExt};
15	use geoengine_datatypes::collections::{
16	BuilderProvider, GeoFeatureCollectionRowBuilder, MultiPointCollection, VectorDataType,
17	};
18	use geoengine_datatypes::dataset::NamedData;
19	use geoengine_datatypes::primitives::SpatialResolution;
20	use geoengine_datatypes::{
21	primitives::{
22	BoundingBox2D, ColumnSelection, Coordinate2D, TimeInterval, VectorQueryRectangle,
23	},
24	spatial_reference::SpatialReference,
25	};
26	use serde::{Deserialize, Serialize};
27	use snafu::{OptionExt, ResultExt, ensure};
28	use std::path::PathBuf;
29	use std::pin::Pin;
30	use std::sync::atomic::Ordering;
31	use std::sync::{Arc, Mutex};
32	use std::{fs::File, sync::atomic::AtomicBool};
33
34	/// Parameters for the CSV Source Operator
35	#[derive(Debug, PartialEq, Eq, Serialize, Deserialize, Clone)]
36	#[serde(rename_all = "camelCase")]
37	pub struct CsvSourceParameters {
38	pub file_path: PathBuf,
39	pub field_separator: char,
40	pub geometry: CsvGeometrySpecification,
41	#[serde(default)]
42	pub time: CsvTimeSpecification,
43	}
44
45	#[derive(Debug, PartialEq, Eq, Serialize, Deserialize, Clone)]
46	#[serde(tag = "type", rename_all = "lowercase")]
47	pub enum CsvGeometrySpecification {
48	#[allow(clippy::upper_case_acronyms)]
49	XY { x: String, y: String },
50	}
51
52	#[derive(Debug, PartialEq, Eq, Serialize, Deserialize, Clone)]
53	pub enum CsvTimeSpecification {
54	None,
55	}
56
57	impl Default for CsvTimeSpecification {
58	fn default() -> Self {	1✔
59	Self::None	1✔
60	}	1✔
61	}
62
63	enum ReaderState {
64	Untouched(Reader<File>),
65	OnGoing {
66	header: ParsedHeader,
67	records: csv::StringRecordsIntoIter<File>,
68	},
69	Error,
70	}
71
72	impl ReaderState {
73	pub fn setup_once(&mut self, geometry_specification: CsvGeometrySpecification) -> Result<()> {	17✔
74	if let ReaderState::Untouched(..) = self {	17✔
75	// pass	7✔
76	} else {	7✔
77	return Ok(());	10✔
78	}
79
80	let old_state = std::mem::replace(self, ReaderState::Error);	7✔
81
82	if let ReaderState::Untouched(mut csv_reader) = old_state {	7✔
83	let header = CsvSourceStream::setup_read(geometry_specification, &mut csv_reader)?;	7✔
84
85	let mut records = csv_reader.into_records();	6✔
86
87	// consume the first row, which is the header
88	if header.has_header {	6✔
89	// TODO: throw error	6✔
90	records.next();	6✔
91	}	6✔
92
93	*self = ReaderState::OnGoing { header, records }	6✔
94	}	×
95
96	Ok(())	6✔
97	}	17✔
98	}
99
100	pub struct CsvSourceStream {
101	parameters: CsvSourceParameters,
102	bbox: BoundingBox2D,
103	chunk_size: usize,
104	reader_state: Arc<Mutex<ReaderState>>,
105	thread_is_computing: Arc<AtomicBool>,
106	#[allow(clippy::option_option)]
107	poll_result: Arc<Mutex<Option<Option<Result<MultiPointCollection>>>>>,
108	}
109
110	pub type CsvSource = SourceOperator<CsvSourceParameters>;
111
112	impl OperatorName for CsvSource {
113	const TYPE_NAME: &'static str = "CsvSource";
114	}
115
116	impl OperatorData for CsvSourceParameters {
117	fn data_names_collect(&self, _data_names: &mut Vec<NamedData>) {}	×
118	}
119
120	#[typetag::serde]	×
121	#[async_trait]
122	impl VectorOperator for CsvSource {
123	async fn _initialize(
124	self: Box<Self>,
125	path: WorkflowOperatorPath,
126	_context: &dyn crate::engine::ExecutionContext,
127	) -> Result<Box<dyn InitializedVectorOperator>> {	8✔
128	let initialized_source = InitializedCsvSource {	4✔
129	name: CanonicOperatorName::from(&self),	4✔
130	path,	4✔
131	result_descriptor: VectorResultDescriptor {	4✔
132	data_type: VectorDataType::MultiPoint, // TODO: get as user input	4✔
133	spatial_reference: SpatialReference::epsg_4326().into(), // TODO: get as user input	4✔
134	columns: Default::default(), // TODO: get when source allows loading other columns	4✔
135	time: None,	4✔
136	bbox: None,	4✔
137	},	4✔
138	state: self.params,	4✔
139	};	4✔
140
141	Ok(initialized_source.boxed())	4✔
142	}	8✔
143
144	span_fn!(CsvSource);
145	}
146
147	pub struct InitializedCsvSource {
148	name: CanonicOperatorName,
149	path: WorkflowOperatorPath,
150	result_descriptor: VectorResultDescriptor,
151	state: CsvSourceParameters,
152	}
153
154	impl InitializedVectorOperator for InitializedCsvSource {
155	fn query_processor(&self) -> Result<crate::engine::TypedVectorQueryProcessor> {	3✔
156	Ok(TypedVectorQueryProcessor::MultiPoint(	3✔
157	CsvSourceProcessor {	3✔
158	params: self.state.clone(),	3✔
159	result_descriptor: self.result_descriptor.clone(),	3✔
160	}	3✔
161	.boxed(),	3✔
162	))	3✔
163	}	3✔
164
165	fn result_descriptor(&self) -> &VectorResultDescriptor {	4✔
166	&self.result_descriptor	4✔
167	}	4✔
168
169	fn canonic_name(&self) -> CanonicOperatorName {	×
170	self.name.clone()	×
171	}	×
172
173	fn name(&self) -> &'static str {	3✔
174	CsvSource::TYPE_NAME	3✔
175	}	3✔
176
177	fn path(&self) -> WorkflowOperatorPath {	3✔
178	self.path.clone()	3✔
179	}	3✔
180
NEW 181	fn optimize(	×
NEW 182	&self,	×
NEW 183	_target_resolution: SpatialResolution,	×
NEW 184	) -> Result<Box<dyn VectorOperator>, OptimizationError> {	×
NEW 185	Ok(CsvSource {	×
NEW 186	params: self.state.clone(),	×
NEW 187	}	×
NEW 188	.boxed())	×
NEW 189	}	×
190	}
191
192	impl CsvSourceStream {
193	/// Creates a new `CsvSource`
194	///
195	/// # Errors
196	///
197	/// This constructor fails if the delimiter is not an ASCII character.
198	/// Furthermore, there are IO errors from the reader.
199	///
200	// TODO: include time interval, e.g. QueryRectangle parameter
201	pub fn new(	7✔
202	parameters: CsvSourceParameters,	7✔
203	bbox: BoundingBox2D,	7✔
204	chunk_size: usize,	7✔
205	) -> Result<Self> {	7✔
206	ensure!(	7✔
207	parameters.field_separator.is_ascii(),	7✔
208	error::CsvSource {	×
209	details: "Delimiter must be ASCII character"	×
210	}	×
211	);
212
213	Ok(Self {
214	reader_state: Arc::new(Mutex::new(ReaderState::Untouched(	7✔
215	csv::ReaderBuilder::new()	7✔
216	.delimiter(parameters.field_separator as u8)	7✔
217	.has_headers(true)	7✔
218	.from_path(parameters.file_path.as_path())	7✔
219	.context(error::CsvSourceReader {})?,	7✔
220	))),
221	thread_is_computing: Arc::new(AtomicBool::new(false)),	7✔
222	poll_result: Arc::new(Mutex::new(None)),	7✔
223	parameters,	7✔
224	bbox,	7✔
225	chunk_size,	7✔
226	})
227	}	7✔
228
229	fn setup_read(	7✔
230	geometry_specification: CsvGeometrySpecification,	7✔
231	csv_reader: &mut Reader<File>,	7✔
232	) -> Result<ParsedHeader> {	7✔
233	csv_reader	7✔
234	.seek(Position::new())	7✔
235	.context(error::CsvSourceReader {})?; // start at beginning	7✔
236
237	ensure!(	7✔
238	csv_reader.has_headers(),	7✔
239	error::CsvSource {	×
240	details: "CSV file must contain header",	×
241	}	×
242	);
243
244	let header = csv_reader.headers().context(error::CsvSourceReader)?;	7✔
245
246	let CsvGeometrySpecification::XY { x, y } = geometry_specification;	7✔
247	let x_index = header	7✔
248	.iter()	7✔
249	.position(\|v\| v == x)	7✔
250	.context(error::CsvSource {	7✔
251	details: "Cannot find x index in csv header",	7✔
252	})?;	7✔
253	let y_index = header	7✔
254	.iter()	7✔
255	.position(\|v\| v == y)	14✔
256	.context(error::CsvSource {	7✔
257	details: "Cannot find y index in csv header",	7✔
258	})?;	7✔
259
260	Ok(ParsedHeader {	6✔
261	has_header: true,	6✔
262	x_index,	6✔
263	y_index,	6✔
264	})	6✔
265	}	7✔
266
267	/// Parse a single CSV row
268	fn parse_row(header: &ParsedHeader, row: &StringRecord) -> Result<ParsedRow> {	17✔
269	let x: f64 = row	17✔
270	.get(header.x_index)	17✔
271	.context(error::CsvSource {	17✔
272	details: "Cannot find x index key",	17✔
273	})?	17✔
274	.parse()	17✔
275	.map_err(\|_error\| error::Error::CsvSource {	17✔
276	details: "Cannot parse x coordinate".to_string(),	×
277	})?;	×
278	let y: f64 = row	17✔
279	.get(header.y_index)	17✔
280	.context(error::CsvSource {	17✔
281	details: "Cannot find y index key",	17✔
282	})?	17✔
283	.parse()	17✔
284	.map_err(\|_error\| error::Error::CsvSource {	17✔
285	details: "Cannot parse y coordinate".to_string(),	×
286	})?;	×
287
288	Ok(ParsedRow {	17✔
289	coordinate: (x, y).into(),	17✔
290	time_interval: TimeInterval::default(),	17✔
291	})	17✔
292	}	17✔
293	}
294
295	impl Stream for CsvSourceStream {
296	type Item = Result<MultiPointCollection>;
297
298	fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {	34✔
299	// TODO: handle lock poisoning on multiple occasions
300
301	if self.thread_is_computing.load(Ordering::Relaxed) {	34✔
302	return Poll::Pending;	×
303	}	34✔
304
305	let mut poll_result = safe_lock_mutex(&self.poll_result);	34✔
306	if let Some(x) = poll_result.take() {	34✔
307	return Poll::Ready(x);	17✔
308	}	17✔
309
310	self.thread_is_computing.store(true, Ordering::Relaxed);	17✔
311
312	let is_working = self.thread_is_computing.clone();	17✔
313	let reader_state = self.reader_state.clone();	17✔
314	let poll_result = self.poll_result.clone();	17✔
315
316	let bbox = self.bbox;	17✔
317	let chunk_size = self.chunk_size;	17✔
318	let parameters = self.parameters.clone();	17✔
319	let waker = cx.waker().clone();	17✔
320
321	crate::util::spawn_blocking(move \|\| {	17✔
322	let mut csv_reader = safe_lock_mutex(&reader_state);	17✔
323	let computation_result = \|\| -> Result<Option<MultiPointCollection>> {	17✔
324	// TODO: is clone necessary?
325	let geometry_specification = parameters.geometry.clone();	17✔
326	csv_reader.setup_once(geometry_specification)?;	17✔
327
328	let (header, records) = match &mut *csv_reader {	16✔
329	ReaderState::OnGoing { header, records } => (header, records),	15✔
330	ReaderState::Error => return Ok(None),	1✔
331	ReaderState::Untouched(_) => unreachable!(),	×
332	};
333
334	let mut builder = MultiPointCollection::builder().finish_header();	15✔
335	let mut number_of_entries = 0; // TODO: add size/len to builder	15✔
336
337	while number_of_entries < chunk_size {	32✔
338	let Some(record) = records.next() else {	29✔
339	break;	11✔
340	};
341
342	let row = record.context(error::CsvSourceReader)?;	18✔
343	let parsed_row = CsvSourceStream::parse_row(header, &row)?;	17✔
344
345	// TODO: filter time
346	if bbox.contains_coordinate(&parsed_row.coordinate) {	17✔
347	builder.push_geometry(parsed_row.coordinate.into());	16✔
348	builder.push_time_interval(parsed_row.time_interval);	16✔
349	builder.finish_row();	16✔
350		16✔
351	number_of_entries += 1;	16✔
352	}	16✔
353	}
354
355	// TODO: is this the correct cancellation criterion?
356	if number_of_entries > 0 {	14✔
357	let collection = builder.build()?;	8✔
358	Ok(Some(collection))	8✔
359	} else {
360	Ok(None)	6✔
361	}
362	}();	17✔
363
364	*safe_lock_mutex(&poll_result) = Some(match computation_result {	17✔
365	Ok(Some(collection)) => Some(Ok(collection)),	8✔
366	Ok(None) => None,	7✔
367	Err(e) => Some(Err(e)),	2✔
368	});
369	is_working.store(false, Ordering::Relaxed);	17✔
370
371	waker.wake();	17✔
372	});	17✔
373
374	Poll::Pending	17✔
375	}	34✔
376	}
377
378	#[derive(Debug)]
379	struct CsvSourceProcessor {
380	params: CsvSourceParameters,
381	result_descriptor: VectorResultDescriptor,
382	}
383
384	#[async_trait]
385	impl QueryProcessor for CsvSourceProcessor {
386	type Output = MultiPointCollection;
387	type SpatialBounds = BoundingBox2D;
388	type Selection = ColumnSelection;
389	type ResultDescription = VectorResultDescriptor;
390
391	async fn _query<'a>(
392	&'a self,
393	query: VectorQueryRectangle,
394	_ctx: &'a dyn QueryContext,
395	) -> Result<BoxStream<'a, Result<Self::Output>>> {	8✔
396	// TODO: properly handle chunk_size
397	Ok(CsvSourceStream::new(self.params.clone(), query.spatial_bounds(), 10)?.boxed())	4✔
398	}	8✔
399
400	fn result_descriptor(&self) -> &VectorResultDescriptor {	13✔
401	&self.result_descriptor	13✔
402	}	13✔
403	}
404
405	#[derive(Clone, Copy, Debug)]
406	struct ParsedHeader {
407	pub has_header: bool,
408	pub x_index: usize,
409	pub y_index: usize,
410	}
411
412	struct ParsedRow {
413	pub coordinate: Coordinate2D,
414	pub time_interval: TimeInterval,
415	// TODO: fields
416	}
417
418	#[cfg(test)]
419	mod tests {
420	use super::*;
421	use crate::engine::MockExecutionContext;
422	use geoengine_datatypes::{
423	collections::{FeatureCollectionInfos, ToGeoJson},
424	util::test::TestDefault,
425	};
426	use std::io::{Seek, SeekFrom, Write};
427
428	#[test]
429	fn it_deserializes() {	1✔
430	let json_string = r#"	1✔
431	{	1✔
432	"type": "CsvSource",	1✔
433	"params": {	1✔
434	"filePath": "/foo/bar.csv",	1✔
435	"fieldSeparator": ",",	1✔
436	"geometry": {	1✔
437	"type": "xy",	1✔
438	"x": "x",	1✔
439	"y": "y"	1✔
440	}	1✔
441	}	1✔
442	}"#;	1✔
443
444	let operator: CsvSource = serde_json::from_str(json_string).unwrap();	1✔
445
446	assert_eq!(	1✔
447	operator,
448	CsvSource {	1✔
449	params: CsvSourceParameters {	1✔
450	file_path: "/foo/bar.csv".into(),	1✔
451	field_separator: ',',	1✔
452	geometry: CsvGeometrySpecification::XY {	1✔
453	x: "x".into(),	1✔
454	y: "y".into()	1✔
455	},	1✔
456	time: CsvTimeSpecification::None,	1✔
457	},	1✔
458	}	1✔
459	);
460	}	1✔
461
462	#[tokio::test]
463	async fn read_points() {	1✔
464	let mut fake_file = tempfile::NamedTempFile::new().unwrap();	1✔
465	write!(	1✔
466	fake_file,	1✔
467	"\	1✔
468	x,y	1✔
469	0,1	1✔
470	2,3	1✔
471	4,5	1✔
472	"	1✔
473	)
474	.unwrap();	1✔
475	fake_file.seek(SeekFrom::Start(0)).unwrap();	1✔
476
477	let mut csv_source = CsvSourceStream::new(	1✔
478	CsvSourceParameters {	1✔
479	file_path: fake_file.path().into(),	1✔
480	field_separator: ',',	1✔
481	geometry: CsvGeometrySpecification::XY {	1✔
482	x: "x".into(),	1✔
483	y: "y".into(),	1✔
484	},	1✔
485	time: CsvTimeSpecification::None,	1✔
486	},	1✔
487	BoundingBox2D::new_unchecked((0., 0.).into(), (5., 5.).into()),	1✔
488	2,
489	)
490	.unwrap();	1✔
491
492	assert_eq!(csv_source.next().await.unwrap().unwrap().len(), 2);	1✔
493	assert_eq!(csv_source.next().await.unwrap().unwrap().len(), 1);	1✔
494	assert!(csv_source.next().await.is_none());	1✔
495	}	1✔
496
497	#[tokio::test]
498	async fn erroneous_point_rows() {	1✔
499	let mut fake_file = tempfile::NamedTempFile::new().unwrap();	1✔
500	write!(	1✔
501	fake_file,	1✔
502	"\	1✔
503	x,y	1✔
504	0,1	1✔
505	CORRUPT	1✔
506	4,5	1✔
507	"	1✔
508	)
509	.unwrap();	1✔
510	fake_file.seek(SeekFrom::Start(0)).unwrap();	1✔
511
512	let mut csv_source = CsvSourceStream::new(	1✔
513	CsvSourceParameters {	1✔
514	file_path: fake_file.path().into(),	1✔
515	field_separator: ',',	1✔
516	geometry: CsvGeometrySpecification::XY {	1✔
517	x: "x".into(),	1✔
518	y: "y".into(),	1✔
519	},	1✔
520	time: CsvTimeSpecification::None,	1✔
521	},	1✔
522	BoundingBox2D::new_unchecked((0., 0.).into(), (5., 5.).into()),	1✔
523	1,
524	)
525	.unwrap();	1✔
526
527	assert_eq!(csv_source.next().await.unwrap().unwrap().len(), 1);	1✔
528	assert!(csv_source.next().await.unwrap().is_err());	1✔
529	assert_eq!(csv_source.next().await.unwrap().unwrap().len(), 1);	1✔
530	assert!(csv_source.next().await.is_none());	1✔
531	}	1✔
532
533	#[tokio::test]
534	async fn corrupt_point_header() {	1✔
535	let mut fake_file = tempfile::NamedTempFile::new().unwrap();	1✔
536	write!(	1✔
537	fake_file,	1✔
538	"\	1✔
539	x,z	1✔
540	0,1	1✔
541	2,3	1✔
542	4,5	1✔
543	"	1✔
544	)
545	.unwrap();	1✔
546	fake_file.seek(SeekFrom::Start(0)).unwrap();	1✔
547
548	let mut csv_source = CsvSourceStream::new(	1✔
549	CsvSourceParameters {	1✔
550	file_path: fake_file.path().into(),	1✔
551	field_separator: ',',	1✔
552	geometry: CsvGeometrySpecification::XY {	1✔
553	x: "x".into(),	1✔
554	y: "y".into(),	1✔
555	},	1✔
556	time: CsvTimeSpecification::None,	1✔
557	},	1✔
558	BoundingBox2D::new_unchecked((0., 0.).into(), (5., 5.).into()),	1✔
559	1,
560	)
561	.unwrap();	1✔
562
563	assert!(csv_source.next().await.unwrap().is_err());	1✔
564	assert!(csv_source.next().await.is_none());	1✔
565	}	1✔
566
567	#[tokio::test]
568	async fn processor() {	1✔
569	let mut fake_file = tempfile::NamedTempFile::new().unwrap();	1✔
570	write!(	1✔
571	fake_file,	1✔
572	"\	1✔
573	x,y	1✔
574	0,1	1✔
575	2,3	1✔
576	4,5	1✔
577	"	1✔
578	)
579	.unwrap();	1✔
580	fake_file.seek(SeekFrom::Start(0)).unwrap();	1✔
581
582	let params = CsvSourceParameters {	1✔
583	file_path: fake_file.path().into(),	1✔
584	field_separator: ',',	1✔
585	geometry: CsvGeometrySpecification::XY {	1✔
586	x: "x".into(),	1✔
587	y: "y".into(),	1✔
588	},	1✔
589	time: CsvTimeSpecification::None,	1✔
590	};	1✔
591
592	let p = CsvSourceProcessor {	1✔
593	params,	1✔
594	result_descriptor: VectorResultDescriptor {	1✔
595	data_type: VectorDataType::MultiPoint,	1✔
596	spatial_reference: SpatialReference::epsg_4326().into(),	1✔
597	columns: Default::default(),	1✔
598	time: None,	1✔
599	bbox: None,	1✔
600	},	1✔
601	};	1✔
602
603	let query = VectorQueryRectangle::new(	1✔
604	BoundingBox2D::new_unchecked(Coordinate2D::new(0., 0.), Coordinate2D::new(3., 3.)),	1✔
605	TimeInterval::new_unchecked(0, 1),	1✔
606	ColumnSelection::all(),	1✔
607	);
608	let ecx = MockExecutionContext::test_default();	1✔
609	let ctx = ecx.mock_query_context((10 * 8 * 2).into());	1✔
610
611	let r: Vec<Result<MultiPointCollection>> =	1✔
612	p.query(query, &ctx).await.unwrap().collect().await;	1✔
613
614	assert_eq!(r.len(), 1);	1✔
615
616	assert_eq!(	1✔
617	serde_json::from_str::<serde_json::Value>(&r[0].as_ref().unwrap().to_geo_json())	1✔
618	.unwrap(),	1✔
619	serde_json::json!({	1✔
620	"type": "FeatureCollection",	1✔
621	"features": [{	1✔
622	"type": "Feature",	1✔
623	"geometry": {	1✔
624	"type": "Point",	1✔
625	"coordinates": [0.0, 1.0]	1✔
626	},	1✔
627	"properties": {},	1✔
628	"when": {	1✔
629	"start": "-262143-01-01T00:00:00+00:00",	1✔
630	"end": "+262142-12-31T23:59:59.999+00:00",	1✔
631	"type": "Interval"	1✔
632	}	1✔
633	}, {	1✔
634	"type": "Feature",	1✔
635	"geometry": {	1✔
636	"type": "Point",	1✔
637	"coordinates": [2.0, 3.0]	1✔
638	},	1✔
639	"properties": {},	1✔
640	"when": {	1✔
641	"start": "-262143-01-01T00:00:00+00:00",	1✔
642	"end": "+262142-12-31T23:59:59.999+00:00",	1✔
643	"type": "Interval"	1✔
644	}	1✔
645	}]	1✔
646	})	1✔
647	);	1✔
648	}	1✔
649
650	#[test]
651	fn operator() {	1✔
652	let mut temp_file = tempfile::NamedTempFile::new().unwrap();	1✔
653	write!(	1✔
654	temp_file,	1✔
655	"\	1✔
656	x;y	1✔
657	0;1	1✔
658	2;3	1✔
659	4;5	1✔
660	"	1✔
661	)
662	.unwrap();	1✔
663	temp_file.seek(SeekFrom::Start(0)).unwrap();	1✔
664
665	let params = CsvSourceParameters {	1✔
666	file_path: temp_file.path().into(),	1✔
667	field_separator: ';',	1✔
668	geometry: CsvGeometrySpecification::XY {	1✔
669	x: "x".into(),	1✔
670	y: "y".into(),	1✔
671	},	1✔
672	time: CsvTimeSpecification::None,	1✔
673	};	1✔
674
675	let operator = CsvSource { params }.boxed();	1✔
676
677	let operator_json = serde_json::to_value(&operator).unwrap();	1✔
678
679	assert_eq!(	1✔
680	operator_json,
681	serde_json::json!({	1✔
682	"type": "CsvSource",	1✔
683	"params": {	1✔
684	"filePath": temp_file.path(),	1✔
685	"fieldSeparator": ";",	1✔
686	"geometry": {	1✔
687	"type": "xy",	1✔
688	"x": "x",	1✔
689	"y": "y"	1✔
690	},
691	"time": "None"	1✔
692	}
693	})
694	);
695
696	let _operator: Box<dyn VectorOperator> = serde_json::from_value(operator_json).unwrap();	1✔
697	}	1✔
698	}

geo-engine / geoengine / 17044350441

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