OMF Transportation Serialization logic

rust/bambam-omf/Cargo.toml

@@ -28,11 +28,16 @@ routee-compass-powertrain = { workspace = true }
 
 arrow = { workspace = true }
 chrono = { workspace = true }
+clap = { workspace = true }
 csv = { workspace = true }
+env_logger = { workspace = true }
+flate2 = { workspace = true }
 futures = { workspace = true }
 geo = { workspace = true }
 geozero = { workspace = true }
 hex = { workspace = true }
+itertools = { workspace = true }
+kdam = { workspace = true }
 log = { workspace = true }
 object_store = { workspace = true }
 parquet = { workspace = true }
@@ -43,3 +48,4 @@ serde_json = { workspace = true }
 serde_arrow = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
+uom = { workspace = true }

rust/bambam-omf/src/app/mod.rs

@@ -0,0 +1,3 @@
+mod omf_app;
+
+pub use omf_app::OmfApp;

rust/bambam-omf/src/app/omf_app.rs

@@ -0,0 +1,55 @@
+use std::path::Path;
+
+use clap::{Parser, Subcommand};
+use serde::{Deserialize, Serialize};
+
+use crate::{
+    collection::{
+        ObjectStoreSource, OvertureMapsCollectionError, OvertureMapsCollectorConfig,
+        ReleaseVersion, RowFilterConfig, TransportationCollection,
+    },
+    graph::OmfGraphVectorized,
+};
+
+/// command line tool for batch downloading and summarizing of GTFS archives
+#[derive(Parser)]
+#[command(author, version, about, long_about = None)]
+#[command(propagate_version = true)]
+pub struct OmfApp {
+    #[command(subcommand)]
+    pub op: OmfOperation,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, Subcommand)]
+pub enum OmfOperation {
+    /// download all of the GTFS archives
+    Download,
+}
+
+impl OmfOperation {
+    pub fn run(self) -> Result<(), OvertureMapsCollectionError> {
+        match self {
+            OmfOperation::Download => {
+                let collector =
+                    OvertureMapsCollectorConfig::new(ObjectStoreSource::AmazonS3, 128).build()?;
+                let release = ReleaseVersion::Latest;
+                let row_filter_config = RowFilterConfig::Bbox {
+                    xmin: -105.254,
+                    xmax: -105.197,
+                    ymin: 39.733,
+                    ymax: 39.784,
+                };
+
+                let collection = TransportationCollection::try_from_collector(
+                    collector,
+                    release,
+                    Some(row_filter_config),
+                )?;
+                let vectorized_graph = OmfGraphVectorized::try_from_collection(collection, 0)?;
+                vectorized_graph.write_compass(Path::new("./"), true)?;
+
+                Ok(())
+            }
+        }
+    }
+}

rust/bambam-omf/src/app/serialize_options.rs

@@ -0,0 +1,13 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Serialize, Deserialize)]
+struct OvertureSerializeOptions {
+    out_file: String,
+    scope: SerializeScope,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+enum SerializeScope {
+    Complete,
+    Compass,
+}

rust/bambam-omf/src/bin/bambam_omf.rs

@@ -0,0 +1,8 @@
+use bambam_omf::app::OmfApp;
+use clap::Parser;
+
+fn main() {
+    env_logger::init();
+    let args = OmfApp::parse();
+    args.op.run().unwrap()
+}

rust/bambam-omf/src/collection/error.rs

@@ -30,4 +30,12 @@ pub enum OvertureMapsCollectionError {
     GroupMappingError(String),
     #[error("Processing records into opportunities failed: {0}")]
     ProcessingError(String),
+    #[error("Serializing record into compass format failed failed: {0}")]
+    SerializationError(String),
+    #[error("Segment connectors vector is invalid or not specified: {0}")]
+    InvalidSegmentConnectors(String),
+    #[error("Invalid or empty geometry: {0}")]
+    InvalidGeometry(String),
+    #[error("Error writing to csv: {0}")]
+    CsvWriteError(String),
 }

rust/bambam-omf/src/collection/mod.rs

@@ -16,6 +16,9 @@ pub use filter::Bbox;
 pub use filter::RowFilter;
 pub use filter::RowFilterConfig;
 pub use object_source::ObjectStoreSource;
-pub use record::{BuildingsRecord, OvertureRecord, OvertureRecordType, PlacesRecord};
+pub use record::{
+    BuildingsRecord, OvertureRecord, OvertureRecordType, PlacesRecord, TransportationCollection,
+    TransportationConnectorRecord, TransportationSegmentRecord,
+};
 pub use taxonomy::{TaxonomyModel, TaxonomyModelBuilder};
 pub use version::ReleaseVersion;

rust/bambam-omf/src/collection/record/mod.rs

@@ -3,13 +3,15 @@ mod common;
 mod overture_record;
 mod place;
 mod record_type;
+mod transportation_collection;
 mod transportation_connector;
 mod transportation_segment;
 
 pub use building::BuildingsRecord;
 pub use overture_record::OvertureRecord;
 pub use place::PlacesRecord;
 pub use record_type::OvertureRecordType;
+pub use transportation_collection::TransportationCollection;
 pub use transportation_connector::TransportationConnectorRecord;
 pub use transportation_segment::TransportationSegmentRecord;
 

rust/bambam-omf/src/collection/record/transportation_collection.rs

@@ -0,0 +1,58 @@
+use crate::collection::{
+    OvertureMapsCollectionError, OvertureMapsCollector, OvertureRecord, OvertureRecordType,
+    ReleaseVersion, RowFilterConfig, TransportationConnectorRecord, TransportationSegmentRecord,
+};
+
+pub struct TransportationCollection {
+    pub connectors: Vec<TransportationConnectorRecord>,
+    pub segments: Vec<TransportationSegmentRecord>,
+}
+
+impl TransportationCollection {
+    /// Use a pre-built collector and download configuration to
+    /// retrieve connectors and segments for a specified query
+    pub fn try_from_collector(
+        collector: OvertureMapsCollector,
+        release: ReleaseVersion,
+        row_filter_config: Option<RowFilterConfig>,
+    ) -> Result<Self, OvertureMapsCollectionError> {
+        let connectors = collector
+            .collect_from_release(
+                release.clone(),
+                &OvertureRecordType::Connector,
+                row_filter_config.clone(),
+            )?
+            .into_iter()
+            .map(|record| match record {
+                OvertureRecord::Connector(transportation_connector_record) => {
+                    Ok(transportation_connector_record)
+                }
+                _ => Err(OvertureMapsCollectionError::DeserializeTypeError(format!(
+                    "expected connector type, got {record:?}"
+                ))),
+            })
+            .collect::<Result<Vec<TransportationConnectorRecord>, OvertureMapsCollectionError>>()?;
+
+        let segments = collector
+            .collect_from_release(
+                release.clone(),
+                &OvertureRecordType::Segment,
+                row_filter_config.clone(),
+            )?
+            .into_iter()
+            .map(|record| match record {
+                OvertureRecord::Segment(transportation_connector_record) => {
+                    Ok(transportation_connector_record)
+                }
+                _ => Err(OvertureMapsCollectionError::DeserializeTypeError(format!(
+                    "expected segment type, got {record:?}"
+                ))),
+            })
+            .collect::<Result<Vec<TransportationSegmentRecord>, OvertureMapsCollectionError>>()?;
+
+        Ok(Self {
+            connectors,
+            segments,
+        })
+    }
+}

rust/bambam-omf/src/collection/record/transportation_connector.rs

@@ -3,14 +3,15 @@ use crate::collection::{OvertureMapsCollectionError, OvertureRecord};
 use super::deserialize_geometry;
 use super::{OvertureMapsBbox, OvertureMapsSource};
 use geo::Geometry;
+use routee_compass_core::model::network::Vertex;
 use serde::{Deserialize, Serialize};
 
 /// Represents a transportation connector record as defined in the Overture Maps Foundation schema.
 /// This struct contains the fields describing a transportation connector, including its unique
 /// identifier, geometry, bounding box, version, and data sources.
 #[derive(Debug, Serialize, Deserialize)]
 pub struct TransportationConnectorRecord {
-    id: Option<String>,
+    pub id: String,
     #[serde(deserialize_with = "deserialize_geometry")]
     geometry: Option<Geometry>,
     bbox: OvertureMapsBbox,
@@ -30,3 +31,27 @@ impl TryFrom<OvertureRecord> for TransportationConnectorRecord {
         }
     }
 }
+
+impl TransportationConnectorRecord {
+    pub fn get_geometry(&self) -> Option<&Geometry> {
+        self.geometry.as_ref()
+    }
+
+    pub fn try_to_vertex(&self, idx: usize) -> Result<Vertex, OvertureMapsCollectionError> {
+        let geometry =
+            self.get_geometry()
+                .ok_or(OvertureMapsCollectionError::SerializationError(format!(
+                    "Invalid or empty geometry {:?}",
+                    self.get_geometry()
+                )))?;
+
+        let (x, y) = match geometry {
+            Geometry::Point(point) => Ok(point.x_y()),
+            _ => Err(OvertureMapsCollectionError::SerializationError(format!(
+                "Incorrect geometry in ConnectorRecord: {geometry:?}"
+            ))),
+        }?;
+
+        Ok(Vertex::new(idx, x as f32, y as f32))
+    }
+}

rust/bambam-omf/src/collection/record/transportation_segment.rs

@@ -1,4 +1,4 @@
-use geo::Geometry;
+use geo::{Geometry, Haversine, Length};
 use serde::{Deserialize, Serialize};
 
 use crate::collection::{OvertureMapsCollectionError, OvertureRecord};
@@ -12,16 +12,16 @@ use super::{OvertureMapsBbox, OvertureMapsNames, OvertureMapsSource};
 /// and other attributes relevant to routing and mapping.
 #[derive(Debug, Serialize, Deserialize)]
 pub struct TransportationSegmentRecord {
-    id: Option<String>,
+    pub id: String,
     #[serde(deserialize_with = "deserialize_geometry")]
-    geometry: Option<Geometry>,
+    pub geometry: Option<Geometry>,
     bbox: OvertureMapsBbox,
     version: i32,
     sources: Option<Vec<Option<OvertureMapsSource>>>,
     subtype: Option<String>,
     class: Option<String>,
     names: Option<OvertureMapsNames>,
-    connectors: Option<Vec<ConnectorReference>>,
+    pub connectors: Option<Vec<ConnectorReference>>,
     routes: Option<Vec<SegmentRoute>>,
     subclass_rules: Option<Vec<SegmentValueBetween<String>>>,
     access_restrictions: Option<Vec<SegmentAccessRestriction>>,
@@ -49,10 +49,28 @@ impl TryFrom<OvertureRecord> for TransportationSegmentRecord {
     }
 }
 
+impl TransportationSegmentRecord {
+    pub fn get_distance_at(&self, at: f64) -> Result<f64, OvertureMapsCollectionError> {
+        let geometry =
+            self.geometry
+                .as_ref()
+                .ok_or(OvertureMapsCollectionError::InvalidGeometry(
+                    "empty geometry".to_string(),
+                ))?;
+
+        match geometry {
+            Geometry::LineString(line_string) => Ok(Haversine.length(line_string) * at),
+            _ => Err(OvertureMapsCollectionError::InvalidGeometry(format!(
+                "geometry was not a linestring {geometry:?}"
+            ))),
+        }
+    }
+}
+
 #[derive(Debug, Serialize, Deserialize)]
-struct ConnectorReference {
-    connector_id: String,
-    at: f64,
+pub struct ConnectorReference {
+    pub connector_id: String,
+    pub at: f64,
 }
 
 #[derive(Debug, Serialize, Deserialize)]

rust/bambam-omf/src/graph/mod.rs

@@ -0,0 +1,6 @@
+mod omf_graph;
+mod segment_split;
+mod serialize_ops;
+mod vertex_serializable;
+
+pub use omf_graph::OmfGraphVectorized;