diff --git a/rust/Cargo.toml b/rust/Cargo.toml index 254125cf..bf9f4700 100644 --- a/rust/Cargo.toml +++ b/rust/Cargo.toml @@ -38,6 +38,7 @@ downloader = { version = "0.2.8" } env_logger = "0.11.8" flate2 = "1.0" futures = { version = "0.3.31", features = ["executor"] } +gbfs_types = { version = "0.1.6", default-features = false, features = ["reqwest_blocking"] } geo = { version = "0.33.1", features = ["use-serde"] } geo-buffer = "0.2.0" geo-traits = "0.3.0" @@ -60,6 +61,7 @@ ordered-float = { version = "5.1.0", features = ["serde"] } osmio = "0.14.0" osmpbf = "0.3.4" parquet = { version = "=58.0.0", features = ["snap", "async", "object_store"] } +pyo3 = { version = "0.29.0", features = ["extension-module", "serde"] } rand = "0.10.0" rayon = "1.10.0" regex = { version = "1.11.1" } diff --git a/rust/bambam-gbfs/Cargo.toml b/rust/bambam-gbfs/Cargo.toml index 09039bd1..a7747b5b 100644 --- a/rust/bambam-gbfs/Cargo.toml +++ b/rust/bambam-gbfs/Cargo.toml @@ -6,15 +6,25 @@ license = "BSD-3-Clause" description = "GBFS Extensions for The Behavior and Advanced Mobility Big Access Model" [dependencies] +bambam-core = { workspace = true } chrono = { workspace = true } clap = { workspace = true } +csv = { workspace = true } env_logger = { workspace = true } +flate2 = { workspace = true } +futures = { workspace = true } +gbfs_types = { workspace = true } geo = { workspace = true } +geozero = { workspace = true } humantime = { workspace = true } +itertools = { workspace = true } kdam = { workspace = true } log = { workspace = true } +rayon = { workspace = true } reqwest = { workspace = true } routee-compass-core = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } serde_with = { workspace = true } +tokio = { workspace = true } +uom = { workspace = true } diff --git a/rust/bambam-gbfs/src/app/download/download_metadata.rs b/rust/bambam-gbfs/src/app/download/download_metadata.rs new file mode 100644 index 00000000..de6f87b3 --- /dev/null +++ b/rust/bambam-gbfs/src/app/download/download_metadata.rs @@ -0,0 +1,32 @@ +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +#[derive(Deserialize, Clone, Debug)] +pub enum UnversionedGbfsVersion { + #[serde(rename = "2.2")] + V2_2, + #[serde(rename = "2.3")] + V2_3, + #[serde(rename = "3.0")] + V3_0, +} + +impl Serialize for UnversionedGbfsVersion { + fn serialize(&self, serializer: S) -> Result + where + S: serde::Serializer, + { + match self { + UnversionedGbfsVersion::V2_2 => serializer.serialize_str("2.2"), + UnversionedGbfsVersion::V2_3 => serializer.serialize_str("2.3"), + UnversionedGbfsVersion::V3_0 => serializer.serialize_str("3.0"), + } + } +} + +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct UnversionedGbfsMetadata { + pub last_updated: Value, + pub ttl: Value, + pub version: UnversionedGbfsVersion, +} diff --git a/rust/bambam-gbfs/src/app/download/entry_point.rs b/rust/bambam-gbfs/src/app/download/entry_point.rs new file mode 100644 index 00000000..c762e788 --- /dev/null +++ b/rust/bambam-gbfs/src/app/download/entry_point.rs @@ -0,0 +1,22 @@ +use clap::ValueEnum; +use serde::{Deserialize, Serialize}; + +/// target of an initial HTTP call to a GBFS archive. +/// for an explanation, see . +#[derive(Serialize, Deserialize, Debug, Clone, Copy, ValueEnum, PartialEq, Eq, Hash)] +#[serde(rename_all = "snake_case")] +pub enum EntryPoint { + /// manifest.json file + Manifest, + /// gbfs.json file for a specific GBFS version. + Gbfs, +} + +impl std::fmt::Display for EntryPoint { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + EntryPoint::Manifest => write!(f, "manifest"), + EntryPoint::Gbfs => write!(f, "gbfs"), + } + } +} diff --git a/rust/bambam-gbfs/src/app/download/gbfs_record.rs b/rust/bambam-gbfs/src/app/download/gbfs_record.rs new file mode 100644 index 00000000..0eefed2d --- /dev/null +++ b/rust/bambam-gbfs/src/app/download/gbfs_record.rs @@ -0,0 +1,240 @@ +use geo::Geometry; +use geozero::{ToGeo, geojson::GeoJson}; +use itertools::Itertools; +use serde::Serialize; + +use crate::{ + app::download::{GbfsVersion, ZoneConstraints}, + model::gbfs::GbfsZoneRecord, +}; + +pub enum GbfsRecord { + V3_0(super::GbfsV3Import), + V2_3(super::GbfsV2_3Import), + V2_2(super::GbfsV2_2Import), +} + +impl Serialize for GbfsRecord { + fn serialize(&self, serializer: S) -> Result + where + S: serde::Serializer, + { + match self { + GbfsRecord::V3_0(record) => record.serialize(serializer), + GbfsRecord::V2_3(record) => record.serialize(serializer), + GbfsRecord::V2_2(record) => record.serialize(serializer), + } + } +} + +impl GbfsRecord { + /// downloads a dataset from a URL for a given GBFS version. + pub async fn download_from_gbfs_endpoint( + client: &reqwest::Client, + url: &str, + version: GbfsVersion, + ) -> Result { + match version { + GbfsVersion::V3_0 => { + let gbfs = super::gbfs_v3_0::run_v3_0_gbfs(client, url).await?; + Ok(Self::V3_0(gbfs)) + } + GbfsVersion::V2_3 => { + let gbfs = super::gbfs_v2_3::run_v2_3_gbfs(client, url).await?; + Ok(Self::V2_3(gbfs)) + } + GbfsVersion::V2_2 => { + let gbfs = super::gbfs_v2_2::run_v2_2_gbfs(client, url).await?; + Ok(Self::V2_2(gbfs)) + } + } + } + + pub fn no_geofence(&self) -> bool { + match self { + GbfsRecord::V3_0(record) => record.geofence.data.geofencing_zones.features.is_empty(), + GbfsRecord::V2_3(record) => record.geofence.data.geofencing_zones.features.is_empty(), + GbfsRecord::V2_2(record) => record.geofence.data.geofencing_zones.features.is_empty(), + } + } + + pub fn system_id(&self) -> String { + match self { + GbfsRecord::V3_0(record) => record.info.data.system_id.clone(), + GbfsRecord::V2_3(record) => record.info.data.system_id.clone(), + GbfsRecord::V2_2(record) => record.info.data.system_id.clone(), + } + } + + pub fn n_features(&self) -> usize { + match self { + GbfsRecord::V3_0(record) => record.geofence.data.geofencing_zones.features.len(), + GbfsRecord::V2_3(record) => record.geofence.data.geofencing_zones.features.len(), + GbfsRecord::V2_2(record) => record.geofence.data.geofencing_zones.features.len(), + } + } + + /// gets the geometry for a feature. + pub fn get_feature_geometry(&self, idx: usize) -> Result { + match self { + GbfsRecord::V3_0(record) => { + let f = record + .geofence + .data + .geofencing_zones + .features + .get(idx) + .ok_or_else(|| format!("feature index {idx} not found"))?; + let geom_str = serde_json::to_string(f) + .map_err(|e| format!("failure deserializing feature: {e}"))?; + let geojson = GeoJson(&geom_str); + let geometry = geojson + .to_geo() + .map_err(|e| format!("unable to read GeoJSON as MultiPolygon: {e}"))?; + Ok(geometry) + } + GbfsRecord::V2_3(record) => { + let f = record + .geofence + .data + .geofencing_zones + .features + .get(idx) + .ok_or_else(|| format!("feature index {idx} not found"))?; + let geom_str = serde_json::to_string(f) + .map_err(|e| format!("failure deserializing feature: {e}"))?; + let geojson = GeoJson(&geom_str); + let geometry = geojson + .to_geo() + .map_err(|e| format!("unable to read GeoJSON as MultiPolygon: {e}"))?; + Ok(geometry) + } + GbfsRecord::V2_2(record) => { + let f = record + .geofence + .data + .geofencing_zones + .features + .get(idx) + .ok_or_else(|| format!("feature index {idx} not found"))?; + let geom_str = serde_json::to_string(f) + .map_err(|e| format!("failure deserializing feature: {e}"))?; + let geojson = GeoJson(&geom_str); + let geometry = geojson + .to_geo() + .map_err(|e| format!("unable to read GeoJSON as MultiPolygon: {e}"))?; + Ok(geometry) + } + } + } + + /// converts a feature in the GBFS dataset into a [GbfsZoneRecord]. + pub fn get_feature_zone_record(&self, idx: usize) -> Result { + match self { + GbfsRecord::V3_0(gbfs) => { + let system_id = gbfs.info.data.system_id.clone(); + let global_constraints = + ZoneConstraints::from_v3_0(gbfs.geofence.data.global_rules.as_ref()); + let feature = gbfs + .geofence + .data + .geofencing_zones + .features + .get(idx) + .ok_or_else(|| format!("feature index {idx} not found"))?; + let found_constraints: Vec = + match feature.properties.rules.as_ref() { + Some(rs) => rs + .iter() + .filter(|r| r.vehicle_type_ids.is_none()) + .map(|r| r.into()) + .collect_vec(), + None => vec![], + }; + let start = feature.properties.start.clone(); + let end = feature.properties.end.clone(); + // merge global and feature-specific constraints + let zone_constraints = ZoneConstraints::merge_constraints( + &global_constraints, + &found_constraints, + None, + ) + .unwrap_or_else(ZoneConstraints::allow_all); + GbfsZoneRecord::new(system_id, idx, start, end, zone_constraints) + } + GbfsRecord::V2_3(gbfs) => { + let system_id = gbfs.info.data.system_id.clone(); + let global_constraints = vec![]; + let feature = gbfs + .geofence + .data + .geofencing_zones + .features + .get(idx) + .ok_or_else(|| format!("feature index {idx} not found"))?; + let found_constraints: Vec = + match feature.properties.rules.as_ref() { + Some(rs) => rs + .iter() + .filter(|r| r.vehicle_type_id.is_none()) + .map(|r| r.into()) + .collect_vec(), + None => vec![], + }; + let start = process_optional_ts_to_string(feature.properties.start)?; + let end = process_optional_ts_to_string(feature.properties.end)?; + // merge global and feature-specific constraints + let zone_constraints = ZoneConstraints::merge_constraints( + &global_constraints, + &found_constraints, + None, + ) + .unwrap_or_else(ZoneConstraints::allow_all); + GbfsZoneRecord::new(system_id, idx, start, end, zone_constraints) + } + GbfsRecord::V2_2(gbfs) => { + let system_id = gbfs.info.data.system_id.clone(); + let global_constraints = vec![]; + let feature = gbfs + .geofence + .data + .geofencing_zones + .features + .get(idx) + .ok_or_else(|| format!("feature index {idx} not found"))?; + let found_constraints: Vec = + match feature.properties.rules.as_ref() { + Some(rs) => rs + .iter() + .filter(|r| r.vehicle_type_id.is_none()) + .map(|r| r.into()) + .collect_vec(), + None => vec![], + }; + let start = process_optional_ts_to_string(feature.properties.start)?; + let end = process_optional_ts_to_string(feature.properties.end)?; + // merge global and feature-specific constraints + let zone_constraints = ZoneConstraints::merge_constraints( + &global_constraints, + &found_constraints, + None, + ) + .unwrap_or_else(ZoneConstraints::allow_all); + GbfsZoneRecord::new(system_id, idx, start, end, zone_constraints) + } + } + } +} + +fn process_optional_ts_to_string(s: Option) -> Result, String> { + match s { + None => Ok(None), + Some(ts) => timestamp_from_int(ts).map(Some), + } +} + +fn timestamp_from_int(t: i64) -> Result { + chrono::DateTime::from_timestamp(t, 0) + .ok_or_else(|| format!("could not parse timestamp '{t}'")) + .map(|ts| ts.to_rfc3339()) +} diff --git a/rust/bambam-gbfs/src/app/download/gbfs_v2_2.rs b/rust/bambam-gbfs/src/app/download/gbfs_v2_2.rs new file mode 100644 index 00000000..6f28d0dc --- /dev/null +++ b/rust/bambam-gbfs/src/app/download/gbfs_v2_2.rs @@ -0,0 +1,172 @@ +use serde::{Deserialize, Serialize}; + +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct GbfsV2_2Import { + /// contains the system-level metadata including opening hours. + pub info: types::SystemInformationFile, + /// contains the zonal geometries, global, and zone-specific traversal rules for this system. + pub geofence: types::GeofencingZonesFile, +} + +/// runs retrieval at the gbfs.json level, retrieving from a single system. +pub async fn run_v2_2_gbfs(client: &reqwest::Client, url: &str) -> Result { + let gbfs: types::GbfsFile = super::ops::retrieve_file(client, url) + .await + .map_err(|e| format!("while downloading gbfs.json file, {e}"))?; + + let geofencing_zones_url = gbfs + .get_geofencing_zones_url("en") + .ok_or_else(|| format!("feed at {url} does not include geofencing_zones"))?; + let geofence: types::GeofencingZonesFile = + super::ops::retrieve_file(client, &geofencing_zones_url) + .await + .map_err(|e| format!("while attempting HTTP GET '{}': {e}", geofencing_zones_url))?; + + let system_info_url = gbfs + .get_system_information_url("en") + .ok_or_else(|| format!("feed at {url} does not include geofencing_zones"))?; + let info: types::SystemInformationFile = super::ops::retrieve_file(client, &system_info_url) + .await + .map_err(|e| format!("while attempting HTTP GET '{}': {e}", system_info_url))?; + + let result = GbfsV2_2Import { info, geofence }; + + Ok(result) +} + +pub mod types { + //! GBFS 2.2 is not implemented in gbfs_types but it is used by Lime. + use std::collections::HashMap; + + use serde::{Deserialize, Serialize}; + + pub type GbfsFile = GbfsMetadata; + pub type SystemInformationFile = GbfsMetadata; + pub type GeofencingZonesFile = GbfsMetadata; + + #[derive(Serialize, Deserialize, Debug, Clone)] + pub struct GbfsMetadata { + /// Indicates the last time data in the feed was updated. This timestamp represents the publisher's knowledge of the current state of the system at this point in time. + pub last_updated: Timestamp, + /// Number of seconds before the data in the feed will be updated again (0 if the data should always be refreshed). + pub ttl: u32, + /// GBFS version number to which the feed conforms, according to the versioning framework. + pub version: String, + /// Response data. + pub data: T, + } + + pub type Feeds = HashMap; + + impl GbfsMetadata { + pub fn get_geofencing_zones_url(&self, language: &str) -> Option { + self.get_file_url(language, "geofencing_zones") + } + + pub fn get_system_information_url(&self, language: &str) -> Option { + self.get_file_url(language, "system_information") + } + + fn get_file_url(&self, language: &str, feed_type: &str) -> Option { + self.data.get(language).and_then(|feed| { + feed.feeds.iter().find_map(|f| { + if f.name == feed_type { + Some(f.url.clone()) + } else { + None + } + }) + }) + } + } + + #[derive(Serialize, Deserialize, Debug, Clone)] + pub struct GbfsLanguageFeeds { + pub feeds: Vec, + } + + #[derive(Serialize, Deserialize, Debug, Clone)] + pub struct GbfsDataFeed { + /// Key identifying the type of feed this is. The key MUST be the base file name defined in the spec for the corresponding feed type + pub name: FeedType, + /// URL for the feed. Note that the actual feed endpoints (urls) may not be defined in the `file_name.json` format. + /// For example, a valid feed endpoint could end with `station_info` instead of `station_information.json`. + pub url: Url, + } + + #[derive(Serialize, Deserialize, Debug, Clone)] + pub struct SystemInformation { + /// Identifier for this vehicle share system. This should be globally unique (even between different systems). + pub system_id: String, + } + + #[derive(Serialize, Deserialize, Debug, Clone)] + pub struct GeofencingZones { + pub geofencing_zones: GeofenceCollection, + } + + #[derive(Serialize, Deserialize, Debug, Clone)] + pub struct GeofenceCollection { + pub r#type: String, + pub features: Vec, + } + + #[derive(Serialize, Deserialize, Debug, Clone)] + pub struct GeofenceFeature { + pub r#type: String, + pub geometry: GeofenceGeometry, + pub properties: GeofenceProperties, + } + + #[derive(Serialize, Deserialize, Debug, Clone)] + #[serde(tag = "type", content = "coordinates")] + pub enum GeofenceGeometry { + #[serde(rename = "MultiPolygon")] + MultiPolygon(Vec>>>), + #[serde(rename = "Polygon")] + Polygon(Vec>>), + } + + #[derive(Serialize, Deserialize, Debug, Clone)] + pub struct GeofenceProperties { + /// Start time of the geofencing zone. If the geofencing zone is always active, this can be omitted. + pub start: Option, + /// End time of the geofencing zone. If the geofencing zone is always active, this can be omitted. + pub end: Option, + /// Array that contains one object per rule. + pub rules: Option>, + } + + #[derive(Serialize, Deserialize, Debug, Clone)] + pub struct GeofenceRules { + pub vehicle_type_id: Option>, + pub ride_allowed: bool, + pub ride_through_allowed: bool, + pub maximum_speed_kph: Option, + } + + pub type Timestamp = i64; + + /// A fully qualified URL that includes `http://` or `https://`. Any special characters in the URL MUST be correctly escaped. See the following for a description of how to create fully qualified URL values. + pub type Url = String; // url::URL; + + /// An IETF BCP 47 language code. For an introduction to IETF BCP 47, refer to and . Examples: `en` for English, `en-US` + pub type Language = String; + + /// Type of a GBFS feed. + /// Current values are : + /// - `gbfs` for [GbfsFile], + /// - `gbfs_versions` for [GbfsVersionsFile], + /// - `system_information` for [SystemInformationFile], + /// - `vehicle_types` for [VehicleTypesFile], + /// - `station_information` for [StationInformationFile], + /// - `station_status` for [StationStatusFile], + /// - `free_bike_status` for [FreeBikeStatusFile], + /// - `system_hours` for [SystemHoursFile], + /// - `system_calendar` for [SystemCalendarFile], + /// - `system_regions` for [SystemRegionsFile], + /// - `system_pricing_plans` for [SystemPricingPlansFile], + /// - `system_alerts` for [SystemAlertsFile], + /// - `geofencing_zones` for [GeofencingZonesFile] + pub type FeedType = String; +} diff --git a/rust/bambam-gbfs/src/app/download/gbfs_v2_3.rs b/rust/bambam-gbfs/src/app/download/gbfs_v2_3.rs new file mode 100644 index 00000000..9f157cf6 --- /dev/null +++ b/rust/bambam-gbfs/src/app/download/gbfs_v2_3.rs @@ -0,0 +1,67 @@ +use gbfs_types::v2_3::files::{GbfsFile, GeofencingZonesFile, SystemInformationFile}; +use serde::{Deserialize, Serialize}; + +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct GbfsV2_3Import { + /// contains the system-level metadata including opening hours. + pub info: SystemInformationFile, + /// contains the zonal geometries, global, and zone-specific traversal rules for this system. + pub geofence: GeofencingZonesFile, +} + +/// runs retrieval from a manifest file. allows mutli-system downloads. +pub async fn run_v2_3_manifest( + client: &reqwest::Client, + url: &str, +) -> Result, String> { + let manifest: gbfs_types::v3_0::files::ManifestFile = + super::ops::retrieve_file(client, url).await?; + + // find v3 datasets for each system in this manifest and run the inner retrieval function + let mut results = vec![]; + for system in manifest.data.datasets.iter() { + let system_id = system.system_id.clone(); + let v_search = system.versions.iter().find(|v| v.version == "2.3"); + match v_search { + Some(v2_3) => { + let gbfs_url = v2_3.url.value.clone(); + let result = run_v2_3_gbfs(client, &gbfs_url).await?; + results.push(result); + } + None => return Err(format!("in system {system_id} no v2.3 was found.")), + } + } + + Ok(results) +} + +/// runs retrieval at the gbfs.json level, retrieving from a single system. +pub async fn run_v2_3_gbfs(client: &reqwest::Client, url: &str) -> Result { + let gbfs: GbfsFile = super::ops::retrieve_file(client, url).await?; + + let geofencing_zones_url = gbfs + .data + .get_geofencing_zones_url("en") + .ok_or_else(|| format!("feed at {url} does not include geofencing_zones"))?; + let geofence: GeofencingZonesFile = + super::ops::retrieve_file(client, &geofencing_zones_url.value) + .await + .map_err(|e| { + format!( + "while attempting HTTP GET '{}': {e}", + geofencing_zones_url.value + ) + })?; + + let system_info_url = gbfs + .data + .get_system_information_url("en") + .ok_or_else(|| format!("feed at {url} does not include geofencing_zones"))?; + let info: SystemInformationFile = super::ops::retrieve_file(client, &system_info_url.value) + .await + .map_err(|e| format!("while attempting HTTP GET '{}': {e}", system_info_url.value))?; + + let result = GbfsV2_3Import { info, geofence }; + + Ok(result) +} diff --git a/rust/bambam-gbfs/src/app/download/gbfs_v3_0.rs b/rust/bambam-gbfs/src/app/download/gbfs_v3_0.rs new file mode 100644 index 00000000..9e2c5681 --- /dev/null +++ b/rust/bambam-gbfs/src/app/download/gbfs_v3_0.rs @@ -0,0 +1,67 @@ +use gbfs_types::v3_0::files::{GbfsFile, GeofencingZonesFile, SystemInformationFile}; +use serde::{Deserialize, Serialize}; + +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct GbfsV3Import { + /// contains the system-level metadata including opening hours. + pub info: SystemInformationFile, + /// contains the zonal geometries, global, and zone-specific traversal rules for this system. + pub geofence: GeofencingZonesFile, +} + +/// runs retrieval from a manifest file. allows mutli-system downloads. +pub async fn run_v3_0_manifest( + client: &reqwest::Client, + url: &str, +) -> Result, String> { + let manifest: gbfs_types::v3_0::files::ManifestFile = + super::ops::retrieve_file(client, url).await?; + + // find v3 datasets for each system in this manifest and run the inner retrieval function + let mut results = vec![]; + for system in manifest.data.datasets.iter() { + let system_id = system.system_id.clone(); + let v_search = system.versions.iter().find(|v| v.version == "3.0"); + match v_search { + Some(v3) => { + let gbfs_url = v3.url.value.clone(); + let result = run_v3_0_gbfs(client, &gbfs_url).await?; + results.push(result); + } + None => return Err(format!("in system {system_id} no v3.0 was found.")), + } + } + + Ok(results) +} + +/// runs retrieval at the gbfs.json level, retrieving from a single system. +pub async fn run_v3_0_gbfs(client: &reqwest::Client, url: &str) -> Result { + let gbfs: GbfsFile = super::ops::retrieve_file(client, url).await?; + + let geofencing_zones_url = gbfs + .data + .get_geofencing_zones_url() + .ok_or_else(|| format!("feed at {url} does not include geofencing_zones"))?; + let geofence: GeofencingZonesFile = + super::ops::retrieve_file(client, &geofencing_zones_url.value) + .await + .map_err(|e| { + format!( + "while attempting HTTP GET '{}': {e}", + geofencing_zones_url.value + ) + })?; + + let system_info_url = gbfs + .data + .get_system_information_url() + .ok_or_else(|| format!("feed at {url} does not include geofencing_zones"))?; + let info: SystemInformationFile = super::ops::retrieve_file(client, &system_info_url.value) + .await + .map_err(|e| format!("while attempting HTTP GET '{}': {e}", system_info_url.value))?; + + let result = GbfsV3Import { info, geofence }; + + Ok(result) +} diff --git a/rust/bambam-gbfs/src/app/download/gbfs_version.rs b/rust/bambam-gbfs/src/app/download/gbfs_version.rs new file mode 100644 index 00000000..acd1beae --- /dev/null +++ b/rust/bambam-gbfs/src/app/download/gbfs_version.rs @@ -0,0 +1,44 @@ +use std::str::FromStr; + +use serde::{Deserialize, Serialize}; + +/// GBFS version of the targeted archive. only supported versions are included. +#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum GbfsVersion { + #[serde(rename = "3.0")] + V3_0, + #[serde(rename = "2.3")] + V2_3, + #[serde(rename = "2.2")] + V2_2, +} + +impl GbfsVersion { + pub const ALL: [&'static str; 3] = ["3.0", "2.3", "2.2"]; +} + +impl std::fmt::Display for GbfsVersion { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + GbfsVersion::V3_0 => write!(f, "3.0"), + GbfsVersion::V2_3 => write!(f, "2.3"), + GbfsVersion::V2_2 => write!(f, "2.2"), + } + } +} + +impl FromStr for GbfsVersion { + type Err = String; + + fn from_str(s: &str) -> Result { + match s { + "3.0" => Ok(Self::V3_0), + "2.3" => Ok(Self::V2_3), + "2.2" => Ok(Self::V2_2), + _ => Err(format!( + "unknown version '{s}', must be one of [{}]", + Self::ALL.join(", ") + )), + } + } +} diff --git a/rust/bambam-gbfs/src/app/download/mod.rs b/rust/bambam-gbfs/src/app/download/mod.rs index ffd60131..a2e8c8f4 100644 --- a/rust/bambam-gbfs/src/app/download/mod.rs +++ b/rust/bambam-gbfs/src/app/download/mod.rs @@ -1,3 +1,17 @@ -mod run; +mod entry_point; +mod gbfs_record; +mod gbfs_v2_2; +mod gbfs_v2_3; +mod gbfs_v3_0; +mod gbfs_version; +mod zone_constraints; -pub use run::run_gbfs_download; +pub mod download_metadata; +pub mod ops; +pub mod run; +pub use entry_point::EntryPoint; +pub use gbfs_v2_2::GbfsV2_2Import; +pub use gbfs_v2_3::GbfsV2_3Import; +pub use gbfs_v3_0::GbfsV3Import; +pub use gbfs_version::GbfsVersion; +pub use zone_constraints::ZoneConstraints; diff --git a/rust/bambam-gbfs/src/app/download/ops.rs b/rust/bambam-gbfs/src/app/download/ops.rs new file mode 100644 index 00000000..d7d54535 --- /dev/null +++ b/rust/bambam-gbfs/src/app/download/ops.rs @@ -0,0 +1,51 @@ +use reqwest::{Client, IntoUrl}; +use serde::de::DeserializeOwned; + +/// helper function for running a client HTTP GET call to retrieve a JSON object. +pub async fn retrieve_file( + client: &Client, + url: U, +) -> Result { + let response = client + .get(url) + .header("User-Agent", "rust-reqwest") + .send() + .await + .map_err(|e| format!("failed to connect to GBFS URL: {e}"))?; + let status = response.status(); + if status.is_success() { + let t: T = response.json().await.map_err(|e| { + let type_name = std::any::type_name::(); + format!("failed to deserialize {type_name} file from HTTP response: {e}") + })?; + Ok(t) + } else { + Err(format!("client response is {status}")) + } +} + +/// reads URLs from a CSV file at some column name into a vector. +pub fn gather_feeds(csv_file: &str, url_column: &str) -> Result, String> { + let mut reader = csv::ReaderBuilder::default() + .has_headers(true) + .from_path(csv_file) + .map_err(|e| format!("failed to open CSV '{csv_file}': {e}"))?; + let headers = reader + .headers() + .map_err(|e| format!("file '{csv_file}' failed to read headers: {e}"))?; + let (col_idx, _) = headers + .iter() + .enumerate() + .find(|(_, name)| *name == url_column) + .ok_or_else(|| format!("column '{url_column}' not found"))?; + + let mut urls = vec![]; + for (idx, row_result) in reader.into_records().enumerate() { + let row = row_result.map_err(|e| format!("failed to read CSV row {idx}: {e}"))?; + let url = row + .get(col_idx) + .ok_or_else(|| format!("CSV row {idx} missing col {col_idx}"))?; + urls.push(url.to_string()); + } + Ok(urls) +} diff --git a/rust/bambam-gbfs/src/app/download/run.rs b/rust/bambam-gbfs/src/app/download/run.rs index 8f569e00..796ade7e 100644 --- a/rust/bambam-gbfs/src/app/download/run.rs +++ b/rust/bambam-gbfs/src/app/download/run.rs @@ -1,21 +1,295 @@ -use std::path::Path; - -use chrono::TimeDelta; - -/// downloads GBFS data for some duration. aggregates the resulting rows and writes them -/// to files to be consumed by BAMBAM. -/// -/// # Arguments -/// * url - URL to the GBFS dataset -/// * out_dir - output directory to write the processed GBFS data -/// * dur - how long to poll the GBFS API -/// -/// # Result -/// If successful, returns nothing, otherwise an error -pub fn run_gbfs_download(url: &str, out_dir: &Path, dur: &TimeDelta) -> Result<(), String> { - let dur_secs = dur.as_seconds_f64(); - log::debug!( - "run_gbfs_download with url={url}, out_dir={out_dir:?}, duration (seconds)={dur_secs}" +use std::{ + collections::HashSet, + fs::File, + path::Path, + sync::{Arc, Mutex}, +}; + +use csv::QuoteStyle; +use flate2::{Compression, write::GzEncoder}; +use geozero::ToWkt; +use itertools::Itertools; +use kdam::{Bar, BarBuilder, BarExt}; +use tokio::{ + sync::Semaphore, + time::{Duration, Instant}, +}; + +use crate::app::download::{ + EntryPoint, GbfsVersion, gbfs_record::GbfsRecord, gbfs_v2_2, gbfs_v2_3, gbfs_v3_0, +}; + +const GEOMETRIES_FILENAME: &str = "edges-gbfs-geofences-enumerated.txt.gz"; +const RECORDS_FILENAME: &str = "edges-gbfs-records.csv.gz"; +const SYSTEM_IDS_FILENAME: &str = "edges-system-ids.txt.gz"; + +pub async fn download_one( + url: &str, + out_dir: &Path, + version: GbfsVersion, + overwrite: bool, +) -> Result<(), String> { + log::info!("run_gbfs_download with url={url}, out_dir={out_dir:?}, version={version}"); + + // download GBFS dataset + let client = reqwest::Client::new(); + let gbfs = GbfsRecord::download_from_gbfs_endpoint(&client, url, version).await?; + write_edge_list(&[gbfs], out_dir, overwrite).await +} + +/// designed to download from a GBFS system list file such as +/// . +pub async fn batch_download( + urls: &[String], + entry_point: EntryPoint, + out_dir: &Path, + parallelism: Option, + delay_ms: Option, + no_compass: bool, + no_summary: bool, + overwrite: bool, +) -> Result<(), String> { + let par = parallelism.unwrap_or(1); + if par == 0 { + return Err("parallelism must be greater than zero".to_string()); + } + let del = delay_ms.unwrap_or_default(); + let bar: Arc> = Arc::new(Mutex::new( + BarBuilder::default() + .total(urls.len()) + .desc("gbfs urls") + .build() + .map_err(|e| format!("error building progress bar: {e}"))?, + )); + let client = Arc::new(reqwest::Client::new()); + let semaphore = Arc::new(Semaphore::new(par)); + let mut next_start_at = Instant::now(); + let spacing = Duration::from_millis(del); + + log::info!( + "starting calls to download archives via {entry_point} entry point (parallelism={par}, delay={del})" ); - todo!("download + post-processing logic") + let mut set = tokio::task::JoinSet::new(); + for url in urls.iter() { + let client = client.clone(); + let semaphore = semaphore.clone(); + let url: String = url.to_string(); + let inner_bar = bar.clone(); + let start_delay = next_start_at.saturating_duration_since(Instant::now()); + next_start_at += spacing; + + set.spawn(async move { + if let Ok(mut bar) = inner_bar.lock() { + let _ = bar.update(1); + } + + if !start_delay.is_zero() { + tokio::time::sleep(start_delay).await; + } + + let _permit = semaphore + .acquire_owned() + .await + .map_err(|e| format!("failed to acquire concurrency permit: {e}"))?; + + run_gbfs_download(client, &url, entry_point) + .await + .map_err(|e| format!("URL: '{url}' - {e}")) + }); + } + + let mut results = vec![]; + let mut errors: Vec = vec![]; + while let Some(res) = set.join_next().await { + match res { + Ok(Err(e)) => errors.push(e), + Err(e) => errors.push(format!("error on tokio join of task: {e}")), + Ok(Ok(r)) => results.extend(r), + } + } + + if !no_summary { + write_summaries(&results, out_dir)?; + } + + if !no_compass { + write_edge_list(&results, out_dir, overwrite).await?; + } + + if !errors.is_empty() { + for err in errors.iter() { + log::error!("{err}") + } + log::error!("{} calls failed", errors.len()); + } + + Ok(()) +} + +/// writes GbfsArchives as JSON files in a summary/ directory. +pub fn write_summaries(results: &[GbfsRecord], out_dir: &Path) -> Result<(), String> { + let sum_dir = out_dir.join("summary"); + std::fs::create_dir_all(&sum_dir).map_err(|e| { + format!( + "failure writing summary directory path '{}': {e}", + sum_dir.to_string_lossy() + ) + })?; + for result in results.iter() { + let no_features = result.no_geofence(); + if !no_features { + let filename = result.system_id(); + let filepath = sum_dir.join(&filename); + + std::fs::write( + filepath, + serde_json::to_string_pretty(&result).unwrap_or_default(), + ) + .map_err(|e| format!("failure while writing to '{filename}': {e}"))?; + } + } + Ok(()) +} + +/// writes an edge list to an output directory that covers the provided GBFS archives. +/// fails if overwrite = false and output files are found. +pub async fn write_edge_list( + archives: &[GbfsRecord], + out_dir: &Path, + overwrite: bool, +) -> Result<(), String> { + // process into BAMBAM-GBFS edge list format + let mut geometries = vec![]; + let mut zone_records = vec![]; + let mut ids = HashSet::new(); + + for gbfs in archives.iter() { + for i in 0..gbfs.n_features() { + let geometry = gbfs.get_feature_geometry(i)?; + let record = gbfs.get_feature_zone_record(i)?; + let system_id = record.system_id.clone(); + geometries.push(geometry); + zone_records.push(record); + ids.insert(system_id); + } + } + + // write outputs + std::fs::create_dir_all(out_dir) + .map_err(|e| format!("failure creating output directory location: {e}"))?; + let mut geom_writer = create_writer( + out_dir, + GEOMETRIES_FILENAME, + false, + QuoteStyle::Never, + overwrite, + )?; + let mut record_writer = create_writer( + out_dir, + RECORDS_FILENAME, + true, + QuoteStyle::Necessary, + overwrite, + )?; + let mut id_writer = create_writer( + out_dir, + SYSTEM_IDS_FILENAME, + false, + QuoteStyle::Never, + overwrite, + )?; + + for (idx, geom) in geometries.into_iter().enumerate() { + let wkt_string = geom + .to_wkt() + .map_err(|e| format!("failure converting geometry {idx} into WKT: {e}"))?; + + geom_writer + .serialize(&wkt_string) + .map_err(|e| format!("failure writing geometry {idx} to file: {e}"))? + } + + for (idx, record) in zone_records.into_iter().enumerate() { + record_writer + .serialize(&record) + .map_err(|e| format!("failure writing record {idx} to file: {e}"))? + } + + for (idx, id) in ids.into_iter().sorted().enumerate() { + id_writer + .serialize(&id) + .map_err(|e| format!("failure writing {idx}th id {id} to file: {e}"))? + } + + Ok(()) +} + +async fn run_gbfs_download( + client: Arc, + url: &String, + entry_point: EntryPoint, +) -> Result, String> { + let unversioned: super::download_metadata::UnversionedGbfsMetadata = + super::ops::retrieve_file(&client, url).await?; + + let result: Vec = match unversioned.version { + super::download_metadata::UnversionedGbfsVersion::V2_2 => { + let result = match entry_point { + EntryPoint::Manifest => { + return Err("manifest entry point not supported for version 2.2".to_string()); + } + EntryPoint::Gbfs => gbfs_v2_2::run_v2_2_gbfs(&client, url) + .await + .map(|g| vec![g])?, + }; + result.into_iter().map(GbfsRecord::V2_2).collect() + } + super::download_metadata::UnversionedGbfsVersion::V2_3 => { + let result = match entry_point { + EntryPoint::Manifest => gbfs_v2_3::run_v2_3_manifest(&client, url).await?, + EntryPoint::Gbfs => gbfs_v2_3::run_v2_3_gbfs(&client, url) + .await + .map(|g| vec![g])?, + }; + result.into_iter().map(GbfsRecord::V2_3).collect() + } + super::download_metadata::UnversionedGbfsVersion::V3_0 => { + let result = match entry_point { + EntryPoint::Manifest => gbfs_v3_0::run_v3_0_manifest(&client, url).await?, + EntryPoint::Gbfs => gbfs_v3_0::run_v3_0_gbfs(&client, url) + .await + .map(|g| vec![g])?, + }; + result.into_iter().map(GbfsRecord::V3_0).collect() + } + }; + + Ok(result) +} + +/// helper function to build a filewriter for writing either .csv.gz or +/// .txt.gz files for compass datasets while respecting the user's overwrite +/// preferences and properly formatting WKT outputs. +fn create_writer( + directory: &Path, + filename: &str, + has_headers: bool, + quote_style: QuoteStyle, + overwrite: bool, +) -> Result>, String> { + let filepath = directory.join(filename); + if filepath.exists() && !overwrite { + return Err(format!( + "user chose overwrite=false but file {} exists", + filepath.to_string_lossy() + )); + } + let file = File::create(&filepath) + .map_err(|e| format!("failure creating file {}: {e}", filepath.to_string_lossy()))?; + let buffer = GzEncoder::new(file, Compression::default()); + let writer = csv::WriterBuilder::new() + .has_headers(has_headers) + .quote_style(quote_style) + .from_writer(buffer); + Ok(writer) } diff --git a/rust/bambam-gbfs/src/app/download/zone_constraints.rs b/rust/bambam-gbfs/src/app/download/zone_constraints.rs new file mode 100644 index 00000000..578c8dcf --- /dev/null +++ b/rust/bambam-gbfs/src/app/download/zone_constraints.rs @@ -0,0 +1,233 @@ +use itertools::Itertools; +use serde::{Deserialize, Serialize}; + +pub type VehicleTypeId = String; + +/// geofencing_zones "rules" object that contains logical rules. +#[derive(Default, Clone, Debug)] +pub struct ZoneConstraints { + /// Is the ride allowed to start in this zone? + pub ride_start_allowed: Option, + /// Is the ride allowed to end in this zone? + pub ride_end_allowed: Option, + /// Is the ride allowed to travel through this zone? + pub ride_through_allowed: Option, + /// What is the maximum speed allowed, in kilometers per hour? + pub maximum_speed_kph: Option, + /// Can vehicles only be parked at stations defined in [station_information] within this geofence zone? + pub station_parking: Option, + /// Array of IDs of vehicle types for which any restrictions SHOULD be applied. + /// If vehicle type IDs are not specified, then restrictions apply to all vehicle types. + pub vehicle_type_ids: Option>, +} + +impl ZoneConstraints { + /// default behavior if no Rules are encountered for a feature. + pub fn allow_all() -> Self { + Self { + ride_start_allowed: Some(true), + ride_end_allowed: Some(true), + ride_through_allowed: Some(true), + maximum_speed_kph: Some(i32::MAX), + station_parking: Some(true), + vehicle_type_ids: None, + } + } + + /// converts v2_3 Rules into ZoneConstraints for either a zone or for a global ruleset. + pub fn from_v2_3( + rules: Option<&Vec>, + ) -> Vec { + match rules { + Some(rs) => rs.iter().map(|r| r.into()).collect_vec(), + None => vec![], + } + } + + /// converts v3_0 Rules into ZoneConstraints for either a zone or for a global ruleset. + pub fn from_v3_0( + rules: Option<&Vec>, + ) -> Vec { + match rules { + Some(rs) => rs.iter().map(|r| r.into()).collect_vec(), + None => vec![], + } + } + + /// from a list of constraints, merge according to the rules of precedence found at + /// . + /// + /// here, we ensure that: + /// - When multiple rules in the same array apply to a particular vehicle type, per the + /// semantics of the vehicle_type_ids field, then the earlier rule (in order of the JSON file) + /// takes precedence for that vehicle type. + /// - When a polygon and the global_rules field define rules that apply to a particular + /// vehicle type, then the rules from the polygon take precedence for that vehicle type + /// in the area of the polygon. + pub fn merge_constraints( + global_constraints: &[ZoneConstraints], + constraints: &[ZoneConstraints], + for_vehicle_type: Option<&VehicleTypeId>, + ) -> Option { + let iter: Box> = + match (global_constraints, constraints) { + ([], []) => return None, + (glob, []) => Box::new(glob.iter()), + ([], zone) => Box::new(zone.iter()), + (glob, zone) => Box::new(zone.iter().chain(glob.iter())), + }; + + let mut accumulator = Self::default(); + for c in iter { + if matches_accumulator(c, for_vehicle_type) { + accumulator.append(c); + } + } + Some(accumulator) + } + + /// appends the values of another set of constraints onto this one. + /// + /// per documentation: + /// + /// > When multiple rules in the same array apply to a particular vehicle type, + /// > per the semantics of the vehicle_type_ids field, then the earlier rule + /// > (in order of the JSON file) takes precedence for that vehicle type. + /// + /// see + fn append(&mut self, other: &ZoneConstraints) { + self.maximum_speed_kph = + merge_no_overwrite(self.maximum_speed_kph, other.maximum_speed_kph); + self.ride_start_allowed = + merge_no_overwrite(self.ride_start_allowed, other.ride_start_allowed); + self.ride_end_allowed = merge_no_overwrite(self.ride_end_allowed, other.ride_end_allowed); + self.ride_through_allowed = + merge_no_overwrite(self.ride_through_allowed, other.ride_through_allowed); + self.station_parking = merge_no_overwrite(self.station_parking, other.station_parking); + } +} + +/// the subset of VehicleType related to route planning. ignored fields are commented out. +/// taken from gbfs_types::v3_0::files::vehicle_types. +#[allow(unused)] +#[serde_with::skip_serializing_none] +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct VehicleTypeConstraints { + /// Unique identifier of a vehicle type. + pub vehicle_type_id: VehicleTypeId, + // /// The vehicle's general form factor. + // pub form_factor: Option, + // /// The number of riders (driver included) the vehicle can legally accommodate. + // pub rider_capacity: Option, + // /// Cargo volume available in the vehicle, expressed in liters. For cars, it corresponds to the space between the boot floor, including the storage under the hatch, to the rear shelf in the trunk. + // pub cargo_volume_capacity: Option, + // /// The capacity of the vehicle cargo space (excluding passengers), expressed in kilograms. + // pub cargo_load_capacity: Option, + // /// The primary propulsion type of the vehicle. + // pub propulsion_type: Option, + // // /// Vehicle air quality certificate. Official anti-pollution certificate, based on the information on the vehicle's registration certificate, attesting to its level of pollutant emissions based on a defined standard. In Europe, for example, it is the European emission standard. The aim of this measure is to encourage the use of the least polluting vehicles by allowing them to drive during pollution peaks or in low emission zones. + // // pub eco_labels: Option>, + // /// This represents the furthest distance in meters that the vehicle can travel without recharging or refueling when it has the maximum amount of energy potential (for example, a full battery or full tank of gas). + pub max_range_meters: Option, + /// The public name of this vehicle type. + pub name: Option>, + // /// Description of accessories available in the vehicle. These accessories are part of the vehicle and are not supposed to change frequently. + // pub vehicle_accessories: Option>, + // // /// Maximum quantity of CO2, in grams, emitted per kilometer, according to the [WLTP](https://en.wikipedia.org/wiki/Worldwide_Harmonised_Light_Vehicles_Test_Procedure). + // // pub g_CO2_km: Option, + // /// URL to an image that would assist the user in identifying the vehicle (for example, an image of the vehicle or a logo). + // pub vehicle_image: Option, + // /// The name of the vehicle manufacturer. + // pub make: Option>, + // /// The name of the vehicle model. + // pub model: Option>, + // /// The color of the vehicle. + // pub color: Option, + // /// Customer-readable description of the vehicle type outlining special features or how-tos. + // pub description: Option>, + // /// Number of wheels this vehicle type has. + // pub wheel_count: Option, + /// The maximum speed in kilometers per hour this vehicle is permitted to reach in accordance with local permit and regulations. + pub max_permitted_speed: Option, + // /// The rated power of the motor for this vehicle type in watts. + // pub rated_power: Option, + // /// Maximum time in minutes that a vehicle can be reserved before a rental begins. + // /// If default_reserve_time is set to 0, the vehicle type cannot be reserved. + // pub default_reserve_time: Option, + // /// The conditions for returning the vehicle at the end of the rental. + // pub return_constraint: Option, + // pub vehicle_assets: Option, + // /// A plan_id, as defined in system_pricing_plans.json, that identifies a default pricing plan for this vehicle to be used by trip planning applications for purposes of calculating the cost of a single trip using this vehicle type. + // /// This default pricing plan is superseded by `pricing_plan_id` when `pricing_plan_id` is defined in `vehicle_status.json`. + // pub default_pricing_plan_id: Option, + // /// All pricing plan IDs that are applied to this vehicle type. + // pub pricing_plan_ids: Option>, +} + +impl From<&gbfs_types::v2_3::files::geofencing_zones::Rule> for ZoneConstraints { + fn from(value: &gbfs_types::v2_3::files::geofencing_zones::Rule) -> Self { + Self { + vehicle_type_ids: value.vehicle_type_id.clone(), + ride_start_allowed: Some(value.ride_allowed), + ride_end_allowed: Some(value.ride_allowed), + ride_through_allowed: Some(value.ride_through_allowed), + maximum_speed_kph: value.maximum_speed_kph, + station_parking: value.station_parking, + } + } +} + +impl From<&gbfs_types::v3_0::files::geofencing_zones::Rule> for ZoneConstraints { + fn from(value: &gbfs_types::v3_0::files::geofencing_zones::Rule) -> Self { + Self { + vehicle_type_ids: value.vehicle_type_ids.clone(), + ride_start_allowed: Some(value.ride_start_allowed), + ride_end_allowed: Some(value.ride_end_allowed), + ride_through_allowed: Some(value.ride_through_allowed), + maximum_speed_kph: value.maximum_speed_kph, + station_parking: value.station_parking, + } + } +} + +impl From<&super::gbfs_v2_2::types::GeofenceRules> for ZoneConstraints { + fn from(value: &super::gbfs_v2_2::types::GeofenceRules) -> Self { + Self { + vehicle_type_ids: value.vehicle_type_id.clone(), + ride_start_allowed: Some(value.ride_allowed), + ride_end_allowed: Some(value.ride_allowed), + ride_through_allowed: Some(value.ride_through_allowed), + maximum_speed_kph: value.maximum_speed_kph, + station_parking: None, + } + } +} + +/// helper for testing if the accumulator's vehicle type argument matches the constraint set. +fn matches_accumulator( + constraint: &ZoneConstraints, + for_vehicle_type: Option<&VehicleTypeId>, +) -> bool { + match (for_vehicle_type, &constraint.vehicle_type_ids) { + (None, None) => true, + (None, Some(_)) => false, + (Some(_), None) => false, + (Some(acc_type), Some(c_types)) => c_types.contains(acc_type), + } +} + +/// per documentation: +/// +/// > When multiple rules in the same array apply to a particular vehicle type, +/// > per the semantics of the vehicle_type_ids field, then the earlier rule +/// > (in order of the JSON file) takes precedence for that vehicle type. +/// +/// see s +fn merge_no_overwrite(lhs: Option, rhs: Option) -> Option { + match (lhs, rhs) { + (None, None) => None, + (None, Some(r)) => Some(r), + (Some(l), None) => Some(l), + (Some(l), Some(_)) => Some(l), + } +} diff --git a/rust/bambam-gbfs/src/app/gbfs_cli.rs b/rust/bambam-gbfs/src/app/gbfs_cli.rs index 1cf0a2e2..78b8c6cc 100644 --- a/rust/bambam-gbfs/src/app/gbfs_cli.rs +++ b/rust/bambam-gbfs/src/app/gbfs_cli.rs @@ -1,9 +1,10 @@ use std::path::Path; -use chrono::TimeDelta; use clap::{Parser, Subcommand}; use serde::{Deserialize, Serialize}; +use crate::app::download::{EntryPoint, GbfsVersion}; + /// command line tool providing GBFS processing scripts #[derive(Parser)] #[command(author, version, about, long_about = None)] @@ -16,40 +17,105 @@ pub struct GbfsCliArguments { #[derive(Debug, Clone, Serialize, Deserialize, Subcommand)] pub enum GbfsOperation { - /// runs a GBFS download, writing data from some source URL - /// to an output directory. + /// downloads GBFS archives from a CSV. ignores archives missing geofence data. writes + /// each dataset JSON 3.0 object to a file in the out directory. + BatchDownload { + /// a file like + #[arg(long)] + csv_file: String, + /// column name for CSV column containing URLs + #[arg(long)] + csv_column: String, + /// whether we are targeting a manifest.json file or gbfs.json file. + #[arg(long)] + entry_point: EntryPoint, + /// output directory path. + #[arg(short, long, default_value_t = String::from("."))] + output_directory: String, + #[arg(long, default_value = None)] + parallelism: Option, + /// delay between calls to avoid getting rejected by provider rate limits during scraping. + #[arg(long, default_value = None)] + delay: Option, + /// if true, skip the Compass network edge list import + #[arg(long)] + no_compass: bool, + /// if true, skip writing the raw GBFS datasets to JSON files. + #[arg(long)] + no_summary: bool, + /// whether to overwrite the files if they already exist. + #[arg(long)] + overwrite: bool, + }, + /// downloads a GBFS archive from its .gbfs endpoint. Download { /// a GBFS API URL - #[arg(short, long)] + #[arg(long)] gbfs_url: String, /// output directory path. - #[arg(short, long, default_value_t = String::from("."))] + #[arg(long)] output_directory: String, - /// duration to collect data rows. provide in human-readable time values - /// 2m, 30s, 2h, 2days... - #[arg(short, long, value_parser = parse_duration, default_value = "10m")] - collect_duration: TimeDelta, + /// GBFS version number to download. + #[arg(long)] + version: GbfsVersion, + /// whether to overwrite the files if they already exist. + #[arg(long)] + overwrite: bool, }, } impl GbfsOperation { - pub fn run(&self) -> Result<(), String> { + pub async fn run(&self) -> Result<(), String> { match self { + GbfsOperation::BatchDownload { + csv_file, + csv_column, + entry_point, + output_directory, + parallelism, + delay, + no_compass, + no_summary, + overwrite, + } => { + let urls = crate::app::download::ops::gather_feeds(csv_file, csv_column)?; + log::info!("found {} urls", urls.len()); + crate::app::download::run::batch_download( + &urls, + *entry_point, + Path::new(output_directory), + *parallelism, + *delay, + *no_compass, + *no_summary, + *overwrite, + ) + .await + } GbfsOperation::Download { gbfs_url, output_directory, - collect_duration, - } => crate::app::download::run_gbfs_download( - gbfs_url, - Path::new(output_directory), - collect_duration, - ), + version, + overwrite, + } => { + crate::app::download::run::download_one( + gbfs_url, + Path::new(output_directory), + *version, + *overwrite, + ) + .await + } } } } -fn parse_duration(s: &str) -> Result { - let std_duration = - humantime::parse_duration(s).map_err(|e| format!("Invalid duration: {e}"))?; - chrono::TimeDelta::from_std(std_duration).map_err(|e| format!("TimeDelta out of range: {e}")) -} +// fn parse_duration(s: &str) -> Result { +// let std_duration = +// humantime::parse_duration(s).map_err(|e| format!("Invalid duration: {e}"))?; +// chrono::TimeDelta::from_std(std_duration).map_err(|e| format!("TimeDelta out of range: {e}")) +// } + +// fn parse_version(s: &str) -> Result { +// GbfsVersion::from_str(s) +// } diff --git a/rust/bambam-gbfs/src/main.rs b/rust/bambam-gbfs/src/main.rs index acf2772f..e163c893 100644 --- a/rust/bambam-gbfs/src/main.rs +++ b/rust/bambam-gbfs/src/main.rs @@ -1,10 +1,12 @@ use bambam_gbfs::app::GbfsCliArguments; use clap::Parser; -fn main() { +#[tokio::main] +async fn main() { env_logger::init(); let args = GbfsCliArguments::parse(); - match args.op.run() { + + match args.op.run().await { Ok(_) => log::info!("finished."), Err(e) => { log::error!("failed running bambam_gbfs: {e}"); diff --git a/rust/bambam-gbfs/src/model/constraint/boarding/builder.rs b/rust/bambam-gbfs/src/model/constraint/boarding/builder.rs deleted file mode 100644 index 86f5fb1f..00000000 --- a/rust/bambam-gbfs/src/model/constraint/boarding/builder.rs +++ /dev/null @@ -1,24 +0,0 @@ -use std::sync::Arc; - -use routee_compass_core::model::constraint::{ - ConstraintModelBuilder, ConstraintModelError, ConstraintModelService, -}; -use routee_compass_core::util::geo::PolygonalRTree; - -use super::{BoardingConstraintConfig, BoardingConstraintEngine, BoardingConstraintService}; - -pub struct BoardingConstraintBuilder {} - -impl ConstraintModelBuilder for BoardingConstraintBuilder { - fn build( - &self, - parameters: &serde_json::Value, - ) -> Result, ConstraintModelError> { - let config: BoardingConstraintConfig = serde_json::from_value(parameters.clone()) - .map_err(|e| ConstraintModelError::BuildError(e.to_string()))?; - let rtree = PolygonalRTree::new(vec![]).map_err(ConstraintModelError::BuildError)?; - let engine = BoardingConstraintEngine::new(config, rtree); - let service = BoardingConstraintService::new(engine); - Ok(Arc::new(service)) - } -} diff --git a/rust/bambam-gbfs/src/model/constraint/boarding/config.rs b/rust/bambam-gbfs/src/model/constraint/boarding/config.rs deleted file mode 100644 index f10851ef..00000000 --- a/rust/bambam-gbfs/src/model/constraint/boarding/config.rs +++ /dev/null @@ -1,5 +0,0 @@ -use serde::{Deserialize, Serialize}; - -#[derive(Serialize, Deserialize, Clone, Debug)] -#[serde(rename_all = "snake_case")] -pub struct BoardingConstraintConfig {} diff --git a/rust/bambam-gbfs/src/model/constraint/boarding/engine.rs b/rust/bambam-gbfs/src/model/constraint/boarding/engine.rs deleted file mode 100644 index 10621e5f..00000000 --- a/rust/bambam-gbfs/src/model/constraint/boarding/engine.rs +++ /dev/null @@ -1,75 +0,0 @@ -use routee_compass_core::{ - model::{constraint::ConstraintModelError, network::Vertex}, - util::geo::PolygonalRTree, -}; - -use super::BoardingConstraintConfig; - -// whatever the type of BoardingId should be -type BoardingId = String; - -pub struct BoardingConstraintEngine { - pub config: BoardingConstraintConfig, - pub rtree: PolygonalRTree, -} - -impl BoardingConstraintEngine { - pub fn new( - config: BoardingConstraintConfig, - rtree: PolygonalRTree, - ) -> BoardingConstraintEngine { - BoardingConstraintEngine { config, rtree } - } - - pub fn in_geofence( - &self, - vertex: &Vertex, - geofence_id: &str, - ) -> Result { - let pt = geo::Geometry::Point(geo::Point::new(vertex.x(), vertex.y())); - let mut iter = self.rtree.intersection(&pt).map_err(|e| { - ConstraintModelError::ConstraintModelError(format!( - "failure checking geofence for {:?}: {e}", - vertex.coordinate.x_y() - )) - })?; - match iter.next() { - Some(boundary) => { - let result = boundary.data == geofence_id; - Ok(result) - } - _ => Ok(false), - } - } -} - -#[cfg(test)] -mod tests { - use routee_compass_core::{model::network::Vertex, util::geo::PolygonalRTree}; - - use crate::model::constraint::boarding::BoardingConstraintConfig; - - use super::BoardingConstraintEngine; - - #[test] - fn test_in_geofence() { - let config = BoardingConstraintConfig {}; - let polygon = geo::Geometry::Polygon(geo::Polygon::new( - geo::line_string![ - (0.0, 0.0).into(), - (1.0, 0.0).into(), - (1.0, 1.0).into(), - (0.0, 1.0).into(), - (0.0, 0.0).into() - ], - vec![], - )); - let rtree = PolygonalRTree::new(vec![(polygon, "zone 1".to_string())]) - .expect("test invariant failed: could not build Rtree"); - let engine = BoardingConstraintEngine::new(config, rtree); - let vertex = Vertex::new(0, 0.5, 0.5); - let result = engine.in_geofence(&vertex, "zone 1"); - assert!(result.is_ok()); - assert!(result.unwrap()); - } -} diff --git a/rust/bambam-gbfs/src/model/constraint/boarding/mod.rs b/rust/bambam-gbfs/src/model/constraint/boarding/mod.rs deleted file mode 100644 index 4dc7b8bb..00000000 --- a/rust/bambam-gbfs/src/model/constraint/boarding/mod.rs +++ /dev/null @@ -1,11 +0,0 @@ -mod builder; -mod config; -mod engine; -mod model; -mod service; - -pub use builder::BoardingConstraintBuilder; -pub use config::BoardingConstraintConfig; -pub use engine::BoardingConstraintEngine; -pub use model::BoardingConstraintModel; -pub use service::BoardingConstraintService; diff --git a/rust/bambam-gbfs/src/model/constraint/boarding/model.rs b/rust/bambam-gbfs/src/model/constraint/boarding/model.rs deleted file mode 100644 index cc393505..00000000 --- a/rust/bambam-gbfs/src/model/constraint/boarding/model.rs +++ /dev/null @@ -1,38 +0,0 @@ -use std::sync::Arc; - -use routee_compass_core::model::{ - constraint::ConstraintModel, - state::{StateModel, StateVariable}, - traversal::EdgeFrontierContext, -}; - -use super::BoardingConstraintEngine; - -pub struct BoardingConstraintModel { - pub engine: Arc, -} - -/// restricts where GBFS boarding can occur by zone -impl BoardingConstraintModel { - pub fn new(engine: Arc) -> BoardingConstraintModel { - BoardingConstraintModel { engine } - } -} - -impl ConstraintModel for BoardingConstraintModel { - fn valid_frontier( - &self, - _ctx: &EdgeFrontierContext, - _state: &[StateVariable], - _state_model: &StateModel, - ) -> Result { - todo!() - } - - fn valid_edge( - &self, - _edge: &routee_compass_core::model::network::Edge, - ) -> Result { - todo!() - } -} diff --git a/rust/bambam-gbfs/src/model/constraint/boarding/service.rs b/rust/bambam-gbfs/src/model/constraint/boarding/service.rs deleted file mode 100644 index 6622eb8d..00000000 --- a/rust/bambam-gbfs/src/model/constraint/boarding/service.rs +++ /dev/null @@ -1,30 +0,0 @@ -use std::sync::Arc; - -use super::{BoardingConstraintEngine, BoardingConstraintModel}; - -use routee_compass_core::model::{ - constraint::{ConstraintModel, ConstraintModelError, ConstraintModelService}, - state::StateModel, -}; - -pub struct BoardingConstraintService { - pub engine: Arc, -} - -impl BoardingConstraintService { - pub fn new(engine: BoardingConstraintEngine) -> BoardingConstraintService { - BoardingConstraintService { - engine: Arc::new(engine), - } - } -} - -impl ConstraintModelService for BoardingConstraintService { - fn build( - &self, - _query: &serde_json::Value, - _state_model: Arc, - ) -> Result, ConstraintModelError> { - Ok(Arc::new(BoardingConstraintModel::new(self.engine.clone()))) - } -} diff --git a/rust/bambam-gbfs/src/model/constraint/gbfs_constraint/builder.rs b/rust/bambam-gbfs/src/model/constraint/gbfs_constraint/builder.rs new file mode 100644 index 00000000..15f66a15 --- /dev/null +++ b/rust/bambam-gbfs/src/model/constraint/gbfs_constraint/builder.rs @@ -0,0 +1,28 @@ +use std::sync::Arc; + +use super::{GbfsConstraintConfig, GbfsConstraintEngine, GbfsConstraintService}; + +use routee_compass_core::model::constraint::{ + ConstraintModelBuilder, ConstraintModelError, ConstraintModelService, +}; + +pub struct GbfsConstraintBuilder {} + +impl ConstraintModelBuilder for GbfsConstraintBuilder { + fn build( + &self, + value: &serde_json::Value, + ) -> Result, ConstraintModelError> { + let config: GbfsConstraintConfig = serde_json::from_value(value.clone()).map_err(|e| { + let msg = format!("failure reading config for GbfsConstraint builder: {e}"); + ConstraintModelError::BuildError(msg) + })?; + let engine = GbfsConstraintEngine::try_from(config).map_err(|e| { + let msg = + format!("failure building engine from config for GbfsConstraint builder: {e}"); + ConstraintModelError::BuildError(msg) + })?; + let service = GbfsConstraintService::new(engine); + Ok(Arc::new(service)) + } +} diff --git a/rust/bambam-gbfs/src/model/constraint/gbfs_constraint/config.rs b/rust/bambam-gbfs/src/model/constraint/gbfs_constraint/config.rs new file mode 100644 index 00000000..e6c2596a --- /dev/null +++ b/rust/bambam-gbfs/src/model/constraint/gbfs_constraint/config.rs @@ -0,0 +1,16 @@ +use serde::{Deserialize, Serialize}; + +#[derive(Deserialize, Serialize, Clone, Debug)] +pub struct GbfsConstraintConfig { + /// output of bambam-gbfs CLI import process, contains a record of the + /// identifier, optional start/end times for service, and traversal ruleset + /// for default vehicles (trips without a VehicleTripId). + /// + /// see [crate::model::gbfs::GbfsZoneRecord] + pub zone_record_input_file: String, + /// output of bambam-gbfs CLI import process, contains zonal geometries + /// with matching indices to the zones input file. + pub zone_geometry_input_file: String, + /// system identifiers for all records. these are sorted lexicagraphically. + pub system_ids_input_file: String, +} diff --git a/rust/bambam-gbfs/src/model/constraint/gbfs_constraint/engine.rs b/rust/bambam-gbfs/src/model/constraint/gbfs_constraint/engine.rs new file mode 100644 index 00000000..560d25c4 --- /dev/null +++ b/rust/bambam-gbfs/src/model/constraint/gbfs_constraint/engine.rs @@ -0,0 +1,78 @@ +use std::path::Path; + +use crate::model::{feature, gbfs::GbfsLookupModel}; + +use super::GbfsConstraintConfig; + +use bambam_core::model::state::CategoricalStateMapping; +use chrono::{DateTime, Utc}; +use routee_compass_core::model::{ + constraint::ConstraintModelError, + network::Vertex, + state::{StateModel, StateVariable}, +}; + +pub struct GbfsConstraintEngine { + lookup: GbfsLookupModel, + mapping: CategoricalStateMapping, +} + +impl GbfsConstraintEngine { + /// tests: + /// - are we NOT in a zone? FALSE + /// - have we boarded? + /// - if FALSE, also check if `ride_start_allowed` + /// - does that zone support `ride_through_allowed`? TRUE + /// - otherwise FALSE + pub fn check_valid( + &self, + vertex: &Vertex, + state: &[StateVariable], + state_model: &StateModel, + start_time: DateTime, + ) -> Result { + let service_opt = feature::state::get_system_id(state, state_model, &self.mapping) + .map_err(|e| { + let msg = format!("failure inspecting service id of search state: {e}"); + ConstraintModelError::ConstraintModelError(msg) + })?; + let zones = self + .lookup + .matching_zones(vertex, state, state_model, start_time, service_opt) + .map_err(|e| { + let msg = format!("failure running GBFS rule lookup: {e}"); + ConstraintModelError::ConstraintModelError(msg) + })?; + + let valid = zones + .iter() + .any(|z| z.ride_through_allowed && (service_opt.is_some() || z.ride_start_allowed)); + + Ok(valid) + } +} + +impl TryFrom for GbfsConstraintEngine { + type Error = ConstraintModelError; + + fn try_from(config: GbfsConstraintConfig) -> Result { + let lookup = GbfsLookupModel::new( + &config.zone_record_input_file, + &config.zone_geometry_input_file, + ) + .map_err(|e| { + let msg = format!("failure building GBFS lookup model: {e}"); + ConstraintModelError::BuildError(msg) + })?; + let mapping = CategoricalStateMapping::from_enumerated_category_file(Path::new( + &config.system_ids_input_file, + )) + .map_err(|e| { + ConstraintModelError::BuildError(format!( + "failure while building categorical mapping from {}: {e}", + config.system_ids_input_file + )) + })?; + Ok(Self { lookup, mapping }) + } +} diff --git a/rust/bambam-gbfs/src/model/constraint/gbfs_constraint/mod.rs b/rust/bambam-gbfs/src/model/constraint/gbfs_constraint/mod.rs new file mode 100644 index 00000000..1e278038 --- /dev/null +++ b/rust/bambam-gbfs/src/model/constraint/gbfs_constraint/mod.rs @@ -0,0 +1,19 @@ +//! GbfsConstraint Constraint Model +//! +//! A stubbed version of a constraint model module that compiles. Used in codegen. +//! If code changes in Compass lead to compiler errors in this module, the changes +//! should get updated. + +mod builder; +mod config; +mod engine; +mod model; +mod params; +mod service; + +pub use builder::GbfsConstraintBuilder; +pub use config::GbfsConstraintConfig; +pub use engine::GbfsConstraintEngine; +pub use model::GbfsConstraintModel; +pub use params::GbfsConstraintParams; +pub use service::GbfsConstraintService; diff --git a/rust/bambam-gbfs/src/model/constraint/gbfs_constraint/model.rs b/rust/bambam-gbfs/src/model/constraint/gbfs_constraint/model.rs new file mode 100644 index 00000000..37477e6e --- /dev/null +++ b/rust/bambam-gbfs/src/model/constraint/gbfs_constraint/model.rs @@ -0,0 +1,43 @@ +use std::sync::Arc; + +use super::{GbfsConstraintEngine, GbfsConstraintParams}; + +use routee_compass_core::model::{ + constraint::{ConstraintModel, ConstraintModelError}, + network::Edge, + state::{StateModel, StateVariable}, + traversal::EdgeFrontierContext, +}; + +pub struct GbfsConstraintModel { + pub engine: Arc, + pub params: GbfsConstraintParams, +} + +impl GbfsConstraintModel { + pub fn new(engine: Arc, params: GbfsConstraintParams) -> Self { + // modify this and the struct definition if additional pre-processing + // is required during model instantiation from query parameters. + Self { engine, params } + } +} + +impl ConstraintModel for GbfsConstraintModel { + fn valid_frontier( + &self, + ctx: &EdgeFrontierContext, + state: &[StateVariable], + state_model: &StateModel, + ) -> Result { + self.engine + .check_valid(ctx.dst, state, state_model, self.params.start_time) + .map_err(|e| { + let msg = format!("failure running GBFS constraint model: {e}"); + ConstraintModelError::ConstraintModelError(msg) + }) + } + + fn valid_edge(&self, _edge: &Edge) -> Result { + Ok(true) + } +} diff --git a/rust/bambam-gbfs/src/model/constraint/gbfs_constraint/params.rs b/rust/bambam-gbfs/src/model/constraint/gbfs_constraint/params.rs new file mode 100644 index 00000000..781c82ac --- /dev/null +++ b/rust/bambam-gbfs/src/model/constraint/gbfs_constraint/params.rs @@ -0,0 +1,8 @@ +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; + +#[derive(Deserialize, Serialize, Clone, Debug)] +pub struct GbfsConstraintParams { + /// time the trip starts + pub start_time: DateTime, +} diff --git a/rust/bambam-gbfs/src/model/constraint/gbfs_constraint/service.rs b/rust/bambam-gbfs/src/model/constraint/gbfs_constraint/service.rs new file mode 100644 index 00000000..dad63a13 --- /dev/null +++ b/rust/bambam-gbfs/src/model/constraint/gbfs_constraint/service.rs @@ -0,0 +1,35 @@ +use std::sync::Arc; + +use super::{GbfsConstraintEngine, GbfsConstraintModel, GbfsConstraintParams}; + +use routee_compass_core::model::{ + constraint::{ConstraintModel, ConstraintModelError, ConstraintModelService}, + state::StateModel, +}; + +pub struct GbfsConstraintService { + engine: Arc, +} + +impl GbfsConstraintService { + pub fn new(engine: GbfsConstraintEngine) -> Self { + Self { + engine: Arc::new(engine), + } + } +} + +impl ConstraintModelService for GbfsConstraintService { + fn build( + &self, + query: &serde_json::Value, + #[allow(unused)] state_model: Arc, + ) -> Result, ConstraintModelError> { + let params: GbfsConstraintParams = serde_json::from_value(query.clone()).map_err(|e| { + let msg = format!("failure reading params for GbfsConstraint service: {e}"); + ConstraintModelError::BuildError(msg) + })?; + let model = GbfsConstraintModel::new(self.engine.clone(), params); + Ok(Arc::new(model)) + } +} diff --git a/rust/bambam-gbfs/src/model/constraint/geofence/builder.rs b/rust/bambam-gbfs/src/model/constraint/geofence/builder.rs deleted file mode 100644 index cab1db33..00000000 --- a/rust/bambam-gbfs/src/model/constraint/geofence/builder.rs +++ /dev/null @@ -1,24 +0,0 @@ -use std::sync::Arc; - -use crate::model::constraint::geofence::GeofenceConstraintEngine; - -use super::{GeofenceConstraintConfig, GeofenceConstraintService}; -use routee_compass_core::model::constraint::{ - ConstraintModelBuilder, ConstraintModelError, ConstraintModelService, -}; -use routee_compass_core::util::geo::PolygonalRTree; -pub struct GeofenceConstraintBuilder {} - -impl ConstraintModelBuilder for GeofenceConstraintBuilder { - fn build( - &self, - parameters: &serde_json::Value, - ) -> Result, ConstraintModelError> { - let config: GeofenceConstraintConfig = serde_json::from_value(parameters.clone()) - .map_err(|e| ConstraintModelError::BuildError(e.to_string()))?; - let rtree = PolygonalRTree::new(vec![]).map_err(ConstraintModelError::BuildError)?; - let engine = GeofenceConstraintEngine::new(config, rtree); - let service = GeofenceConstraintService::new(engine); - Ok(Arc::new(service)) - } -} diff --git a/rust/bambam-gbfs/src/model/constraint/geofence/config.rs b/rust/bambam-gbfs/src/model/constraint/geofence/config.rs deleted file mode 100644 index d15558b6..00000000 --- a/rust/bambam-gbfs/src/model/constraint/geofence/config.rs +++ /dev/null @@ -1,5 +0,0 @@ -use serde::{Deserialize, Serialize}; - -#[derive(Serialize, Deserialize, Clone, Debug)] -#[serde(rename_all = "snake_case")] -pub struct GeofenceConstraintConfig {} diff --git a/rust/bambam-gbfs/src/model/constraint/geofence/engine.rs b/rust/bambam-gbfs/src/model/constraint/geofence/engine.rs deleted file mode 100644 index 1db7f901..00000000 --- a/rust/bambam-gbfs/src/model/constraint/geofence/engine.rs +++ /dev/null @@ -1,75 +0,0 @@ -use routee_compass_core::{ - model::{constraint::ConstraintModelError, network::Vertex}, - util::geo::PolygonalRTree, -}; - -use crate::model::constraint::geofence::GeofenceConstraintConfig; - -// whatever the type of GeofenceId should be -type GeofenceId = String; - -pub struct GeofenceConstraintEngine { - pub config: GeofenceConstraintConfig, - pub rtree: PolygonalRTree, -} - -impl GeofenceConstraintEngine { - pub fn new( - config: GeofenceConstraintConfig, - rtree: PolygonalRTree, - ) -> GeofenceConstraintEngine { - GeofenceConstraintEngine { config, rtree } - } - - pub fn in_geofence( - &self, - vertex: &Vertex, - geofence_id: &str, - ) -> Result { - let pt = geo::Geometry::Point(geo::Point::new(vertex.x(), vertex.y())); - let mut iter = self.rtree.intersection(&pt).map_err(|e| { - ConstraintModelError::ConstraintModelError(format!( - "failure checking geofence for {:?}: {e}", - vertex.coordinate.x_y() - )) - })?; - match iter.next() { - Some(boundary) => { - let result = boundary.data == geofence_id; - Ok(result) - } - _ => Ok(false), - } - } -} - -#[cfg(test)] -mod tests { - use routee_compass_core::{model::network::Vertex, util::geo::PolygonalRTree}; - - use crate::model::constraint::geofence::GeofenceConstraintConfig; - - use super::GeofenceConstraintEngine; - - #[test] - fn test_in_geofence() { - let config = GeofenceConstraintConfig {}; - let polygon = geo::Geometry::Polygon(geo::Polygon::new( - geo::line_string![ - (0.0, 0.0).into(), - (1.0, 0.0).into(), - (1.0, 1.0).into(), - (0.0, 1.0).into(), - (0.0, 0.0).into() - ], - vec![], - )); - let rtree = PolygonalRTree::new(vec![(polygon, "zone 1".to_string())]) - .expect("test invariant failed: could not build Rtree"); - let engine = GeofenceConstraintEngine::new(config, rtree); - let vertex = Vertex::new(0, 0.5, 0.5); - let result = engine.in_geofence(&vertex, "zone 1"); - assert!(result.is_ok()); - assert!(result.unwrap()); - } -} diff --git a/rust/bambam-gbfs/src/model/constraint/geofence/mod.rs b/rust/bambam-gbfs/src/model/constraint/geofence/mod.rs deleted file mode 100644 index 190edd4a..00000000 --- a/rust/bambam-gbfs/src/model/constraint/geofence/mod.rs +++ /dev/null @@ -1,11 +0,0 @@ -mod builder; -mod config; -mod engine; -mod model; -mod service; - -pub use builder::GeofenceConstraintBuilder; -pub use config::GeofenceConstraintConfig; -pub use engine::GeofenceConstraintEngine; -pub use model::GeofenceConstraintModel; -pub use service::GeofenceConstraintService; diff --git a/rust/bambam-gbfs/src/model/constraint/geofence/model.rs b/rust/bambam-gbfs/src/model/constraint/geofence/model.rs deleted file mode 100644 index 92227a76..00000000 --- a/rust/bambam-gbfs/src/model/constraint/geofence/model.rs +++ /dev/null @@ -1,39 +0,0 @@ -use std::sync::Arc; - -use routee_compass_core::model::{ - constraint::ConstraintModel, - state::{StateModel, StateVariable}, - traversal::EdgeFrontierContext, -}; - -use crate::model::constraint::geofence::GeofenceConstraintEngine; - -/// looks up a geofence by agency id to test whether an edge traversal -/// does not exit the region supported by this GBFS travel mode. -pub struct GeofenceConstraintModel { - pub engine: Arc, -} - -impl GeofenceConstraintModel { - pub fn new(engine: Arc) -> GeofenceConstraintModel { - GeofenceConstraintModel { engine } - } -} - -impl ConstraintModel for GeofenceConstraintModel { - fn valid_frontier( - &self, - _ctx: &EdgeFrontierContext, - _state: &[StateVariable], - _state_model: &StateModel, - ) -> Result { - todo!() - } - - fn valid_edge( - &self, - _edge: &routee_compass_core::model::network::Edge, - ) -> Result { - todo!() - } -} diff --git a/rust/bambam-gbfs/src/model/constraint/geofence/service.rs b/rust/bambam-gbfs/src/model/constraint/geofence/service.rs deleted file mode 100644 index 4fe607d2..00000000 --- a/rust/bambam-gbfs/src/model/constraint/geofence/service.rs +++ /dev/null @@ -1,30 +0,0 @@ -use std::sync::Arc; - -use crate::model::constraint::geofence::{GeofenceConstraintEngine, GeofenceConstraintModel}; - -use routee_compass_core::model::{ - constraint::{ConstraintModel, ConstraintModelError, ConstraintModelService}, - state::StateModel, -}; - -pub struct GeofenceConstraintService { - pub engine: Arc, -} - -impl GeofenceConstraintService { - pub fn new(engine: GeofenceConstraintEngine) -> GeofenceConstraintService { - GeofenceConstraintService { - engine: Arc::new(engine), - } - } -} - -impl ConstraintModelService for GeofenceConstraintService { - fn build( - &self, - _query: &serde_json::Value, - _state_model: Arc, - ) -> Result, ConstraintModelError> { - Ok(Arc::new(GeofenceConstraintModel::new(self.engine.clone()))) - } -} diff --git a/rust/bambam-gbfs/src/model/constraint/mod.rs b/rust/bambam-gbfs/src/model/constraint/mod.rs index a4435e23..db64a87c 100644 --- a/rust/bambam-gbfs/src/model/constraint/mod.rs +++ b/rust/bambam-gbfs/src/model/constraint/mod.rs @@ -1,2 +1 @@ -pub mod boarding; -pub mod geofence; +pub mod gbfs_constraint; diff --git a/rust/bambam-gbfs/src/model/feature.rs b/rust/bambam-gbfs/src/model/feature.rs new file mode 100644 index 00000000..0d5eba58 --- /dev/null +++ b/rust/bambam-gbfs/src/model/feature.rs @@ -0,0 +1,110 @@ +pub mod fieldname { + /// the name of the agency providing the GBFS vehicle. + /// if this value is set, the trip has boarded the service. + pub const GBFS_SYSTEM_ID: &str = "gbfs_system_id"; + + /// true if the trip has a [GBFS_AGENCY_ID] and if the current + /// edge has a GBFS zone where `ride_end_allowed` is true. + pub const GBFS_DESTINATION: &str = "gbfs_destination"; +} + +pub mod variable { + //! the configuration for state variables in GTFS-Flex routing + + use routee_compass_core::model::state::{CustomVariableConfig, StateVariableConfig}; + + /// stores a zone id in a state variable + pub fn gbfs_system_id() -> StateVariableConfig { + StateVariableConfig::Custom { + custom_type: "Option".to_string(), + value: empty(), + accumulator: true, + } + } + + /// each gbfs destination is assumed to be "true" and conditionally negated by the GBFS traversal model. + pub fn gbfs_destination() -> StateVariableConfig { + StateVariableConfig::Custom { + custom_type: "Bool".to_string(), + value: CustomVariableConfig::Boolean { initial: true }, + accumulator: false, + } + } + + /// empty value is "-1" for categoricals mapped to real numbers + pub fn empty() -> CustomVariableConfig { + CustomVariableConfig::SignedInteger { initial: -1 } + } +} + +pub mod state { + use super::fieldname; + use bambam_core::model::state::CategoricalStateMapping; + use routee_compass_core::model::state::{StateModel, StateModelError, StateVariable}; + + /// label value representing un-assigned agency_ids + const NO_AGENCY_ID: i64 = -1; + + /// assigns the given agency_id to the state vector. + pub fn set_system_id( + state: &mut [StateVariable], + state_model: &StateModel, + system_id: &str, + mapping: &CategoricalStateMapping, + ) -> Result<(), StateModelError> { + let value = mapping.get_label(system_id).ok_or_else(|| { + StateModelError::RuntimeError(format!("system_id {system_id} missing from mapping")) + })?; + state_model.set_custom_i64(state, fieldname::GBFS_SYSTEM_ID, value) + } + + /// gets the stored agency_id from the state variable, if it exists. + pub fn get_system_id<'a, 'b>( + state: &'a [StateVariable], + state_model: &'a StateModel, + mapping: &'b CategoricalStateMapping, + ) -> Result, StateModelError> { + let agency_label = state_model.get_custom_i64(state, fieldname::GBFS_SYSTEM_ID)?; + let agency_id = mapping.get_categorical(agency_label)?; + Ok(agency_id) + } + + /// confirms that there is a stored agency_id and that it matches the provided one. + pub fn verify_system_id( + agency_id: &str, + state: &[StateVariable], + state_model: &StateModel, + mapping: &CategoricalStateMapping, + ) -> Result { + let stored_agency = get_system_id(state, state_model, mapping)?; + match stored_agency { + Some(a) if a == agency_id => Ok(true), + _ => Ok(false), + } + } + + /// confirms that the search has boarded a GBFS agency (that GBFS_AGENCY_ID is set). + pub fn is_boarded( + state: &[StateVariable], + state_model: &StateModel, + ) -> Result { + let agency_id = state_model.get_custom_i64(state, fieldname::GBFS_SYSTEM_ID)?; + Ok(agency_id != NO_AGENCY_ID) + } + + /// affirms that the trip location associated with this state vector is a valid trip destination. + pub fn set_invalid_destination( + state: &mut [StateVariable], + state_model: &StateModel, + ) -> Result<(), StateModelError> { + state_model.set_custom_bool(state, fieldname::GBFS_DESTINATION, &false) + } + + /// is the trip location associated with this state vector is a valid trip destination? + pub fn get_valid_destination( + state: &[StateVariable], + state_model: &StateModel, + ) -> Result { + state_model.get_custom_bool(state, fieldname::GBFS_DESTINATION) + } +} diff --git a/rust/bambam-gbfs/src/model/gbfs/lookup.rs b/rust/bambam-gbfs/src/model/gbfs/lookup.rs new file mode 100644 index 00000000..b0e5d5df --- /dev/null +++ b/rust/bambam-gbfs/src/model/gbfs/lookup.rs @@ -0,0 +1,159 @@ +use crate::model::gbfs::{GbfsZoneRecord, ZoneState}; + +use bambam_core::{model::state::fieldname, util::geo_utils}; +use chrono::{DateTime, TimeDelta, Utc}; +use geo::Geometry; +use geozero::{ToGeo, wkt::Wkt}; +use itertools::Itertools; +use kdam::BarBuilder; +use routee_compass_core::{ + model::{ + network::Vertex, + state::{StateModel, StateVariable}, + }, + util::{ + fs::{read_decoders, read_utils}, + geo::PolygonalRTree, + }, +}; + +pub struct GbfsLookupModel { + rtree: PolygonalRTree, +} + +impl GbfsLookupModel { + pub fn new(zones_input_file: &str, geometries_input_file: &str) -> Result { + let zonal_records = read_records(zones_input_file)?; + let geometries = read_geometries(geometries_input_file)?; + + // check sizes match + if zonal_records.len() != geometries.len() { + let msg = format!( + "file {} has {} records (besides global record), but file {} has {} geometries; sizes must match.", + zones_input_file, + zonal_records.len(), + geometries_input_file, + geometries.len() + ); + return Err(msg); + } + + let rows = geometries.into_iter().zip(zonal_records).collect_vec(); + let rtree = PolygonalRTree::new(rows) + .map_err(|e| format!("failure building spatial index: {e}"))?; + Ok(Self { rtree }) + } + + /// uses the destination vertex of the current edge traversal to find any intersecting + /// zones. filters out zonal rules that do not match our trip datetime. if we are boarded on a GBFS + /// provider we further filter to only zones with a matching ServiceId. combines all + /// zone record rules in order to produce a single value result. + pub fn matching_zones<'b>( + &self, + vertex: &'b Vertex, + state: &'b [StateVariable], + state_model: &'b StateModel, + start_time: DateTime, + service_id: Option<&'b String>, + ) -> Result, String> { + let query = geo::Geometry::Point(geo::Point(vertex.coordinate.0)); + let time = get_trip_datetime(state, state_model, start_time)?; + let intersection = self.spatiotemporal_intersection(&query, time, service_id)?; + let rules = ZoneState::collect_zones(intersection); + Ok(rules) + } + + /// inner function that finds all zone matches that intersect a spatiotemporal query. + fn spatiotemporal_intersection<'a, 'b: 'a>( + &'a self, + spatial_query: &'b Geometry, + time: DateTime, + system_id: Option<&'b String>, + ) -> Result, String> { + let matches = self + .rtree + .intersection(spatial_query) + .map_err(|e| format!("failure running GBFS zone lookup: {e}"))? + .filter_map(move |z| zone_matches_search(&z.data, time, system_id)); + + Ok(matches.collect_vec()) + } +} + +/// times are documented with "If the geofencing zone is always active, this can be omitted." +/// in this way, we optimistically choose `true` when only one of [start,end] are present. +/// +/// if we are currently boarded, we only accept zone records with a matching system id. +fn zone_matches_search<'a>( + record: &'a GbfsZoneRecord, + time: DateTime, + system_id: Option<&String>, +) -> Option<&'a GbfsZoneRecord> { + let service_ok = match system_id { + Some(id) => id == &record.system_id, + None => true, + }; + let time_ok = match (record.start, record.end) { + (Some(s), Some(e)) => s <= time && time < e, + _ => true, + }; + if service_ok && time_ok { + Some(record) + } else { + None + } +} + +/// reads the records and builds a ZoneGraph from them. holds aside the global record +fn read_records(zone_record_input_file: &str) -> Result, String> { + let bb = BarBuilder::default().desc("reading zone records"); + let zone_records: Box<[GbfsZoneRecord]> = + read_utils::from_csv(&zone_record_input_file, true, Some(bb), None) + .map_err(|e| format!("failure reading zone records: {e}"))?; + Ok(zone_records.to_vec()) +} + +/// reads zonal geometries and ZoneIds from a CSV geometry collection. +fn read_geometries(geometry_input_file: &str) -> Result>, String> { + let bb = BarBuilder::default().desc("reading zone geometries"); + let record_strings = + read_utils::read_raw_file(geometry_input_file, read_decoders::string, Some(bb), None) + .map_err(|e| format!("failure reading file '{geometry_input_file}': {e}"))?; + let rtree_data = record_strings + .iter() + .enumerate() + .map(|(idx, geom_str)| { + let geometry = Wkt(geom_str) + .to_geo() + .map_err(|e| format!("failure reading WKT geometry {idx}: {e}"))?; + let geom_f32 = geo_utils::try_convert_f32(&geometry).map_err(|e| { + format!( + "failure converting geometry to 32-bit FP representation for index {idx}: {e}", + ) + })?; + Ok(geom_f32) + }) + .collect::, String>>()?; + + Ok(rtree_data) +} + +/// composes a starting datetime value with the trip time on the provided state producing the +/// current +fn get_trip_datetime( + state: &[StateVariable], + state_model: &StateModel, + start_time: DateTime, +) -> Result, String> { + let time = state_model + .get_time(state, fieldname::TRIP_TIME) + .map_err(|e| format!("failure reading {} from state: {e}", fieldname::TRIP_TIME))?; + let nanos = time.get::() as i64; + let timedelta = TimeDelta::nanoseconds(nanos); + start_time.checked_add_signed(timedelta).ok_or_else(|| { + format!( + "adding {nanos} ns to {} is out-of-range", + start_time.to_rfc3339() + ) + }) +} diff --git a/rust/bambam-gbfs/src/model/gbfs/lookup_config.rs b/rust/bambam-gbfs/src/model/gbfs/lookup_config.rs new file mode 100644 index 00000000..70bb96b0 --- /dev/null +++ b/rust/bambam-gbfs/src/model/gbfs/lookup_config.rs @@ -0,0 +1,14 @@ +use serde::{Deserialize, Serialize}; + +#[derive(Deserialize, Serialize, Clone, Debug)] +pub struct GbfsLookupConfig { + /// output of bambam-gbfs CLI import process, contains a record of the + /// identifier, optional start/end times for service, and traversal ruleset + /// for default vehicles (trips without a VehicleTripId). + /// + /// see [crate::model::gbfs::GbfsZoneRecord] + pub zones_input_file: String, + /// output of bambam-gbfs CLI import process, contains zonal geometries + /// with matching indices to the zones input file. + pub geometries_input_file: String, +} diff --git a/rust/bambam-gbfs/src/model/gbfs/mod.rs b/rust/bambam-gbfs/src/model/gbfs/mod.rs new file mode 100644 index 00000000..421cb5e1 --- /dev/null +++ b/rust/bambam-gbfs/src/model/gbfs/mod.rs @@ -0,0 +1,10 @@ +mod lookup; +mod lookup_config; +mod record; +mod state; + +pub use record::GbfsZoneRecord; +pub mod ops; +pub use lookup::GbfsLookupModel; +pub use lookup_config::GbfsLookupConfig; +pub use state::ZoneState; diff --git a/rust/bambam-gbfs/src/model/gbfs/ops.rs b/rust/bambam-gbfs/src/model/gbfs/ops.rs new file mode 100644 index 00000000..a29bc081 --- /dev/null +++ b/rust/bambam-gbfs/src/model/gbfs/ops.rs @@ -0,0 +1,49 @@ +use chrono::{DateTime, TimeDelta, Utc}; +use routee_compass_core::model::{ + state::{StateModel, StateVariable}, + traversal::default::fieldname, +}; +use uom::si::f64::Time; + +/// builds a globally-unique identifier for a zone. based on the fact that +/// system_ids are defined as globally unique. as documented at +/// : +/// +/// > [system_id] is a globally unique identifier for the vehicle share system. Each distinct system +/// > or geographic area in which vehicles are operated MUST have its own unique system_id. It +/// > is up to the publisher of the feed to guarantee uniqueness and MUST be checked against +/// > existing system_id fields in systems.csv to ensure this. This value is intended to remain +/// > the same over the life of the system. +/// > +/// > System IDs SHOULD be recognizable as belonging to a particular system as opposed to random +/// > strings - for example, bcycle_austin or biketown_pdx. +pub fn fully_qualified_zone_id(system_id: &str, zone_feature_index: usize) -> String { + format!("{system_id}#{zone_feature_index}") +} + +/// helper function to calculate the current datetime based on the start time and trip_time values. +pub fn current_datetime( + start_time: DateTime, + state: &[StateVariable], + state_model: &StateModel, +) -> Result, String> { + let time: Time = state_model + .get_time(state, fieldname::TRIP_TIME) + .map_err(|e| { + format!( + "failure getting '{}' from state vector: {e}", + fieldname::TRIP_TIME + ) + })?; + let time_i64 = time.get::() as i64; + start_time + .checked_add_signed(TimeDelta::seconds(time_i64)) + .ok_or_else(|| { + let msg = format!( + "adding {} seconds to {} was out of bounds", + time_i64, + start_time.to_rfc3339() + ); + msg + }) +} diff --git a/rust/bambam-gbfs/src/model/gbfs/record.rs b/rust/bambam-gbfs/src/model/gbfs/record.rs new file mode 100644 index 00000000..7928ab93 --- /dev/null +++ b/rust/bambam-gbfs/src/model/gbfs/record.rs @@ -0,0 +1,71 @@ +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; + +use crate::{app::download::ZoneConstraints, model::gbfs::ops}; + +/// a composite of SystemInformation and GeofencingZone attributes along with a +/// globally-unique identifier for this zone. the geometry is stored elsewhere. +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct GbfsZoneRecord { + /// a globally-unique zone identifier that combines the system id and zone index + pub fq_id: String, + /// GBFS SystemInformation.system_id value. + pub system_id: String, + /// index of the geojson feature associated with this zone. if not provided, + /// this record is the global zone record. + pub feature_index: usize, + /// optional start time for using this zone + pub start: Option>, + /// optional end time for using this zone + pub end: Option>, + /// Is the ride allowed to start in this zone? + pub ride_start_allowed: bool, + /// Is the ride allowed to end in this zone? + pub ride_end_allowed: bool, + /// Is the ride allowed to travel through this zone? + pub ride_through_allowed: bool, + /// What is the maximum speed allowed, in kilometers per hour? + pub maximum_speed_kph: Option, + /// Can vehicles only be parked at stations defined in [station_information] within this geofence zone? + pub station_parking: bool, +} + +impl GbfsZoneRecord { + /// create a record for a zone, including its identifiers and its constraint set. + /// if a given boolean constraint is found to be None, apply a permissive rule. + pub fn new( + system_id: String, + feature_index: usize, + start: Option, + end: Option, + zone_constraints: ZoneConstraints, + ) -> Result { + let fq_id = ops::fully_qualified_zone_id(&system_id, feature_index); + let start = to_datetime(start)?; + let end = to_datetime(end)?; + let result = Self { + fq_id, + system_id, + feature_index, + start, + end, + ride_start_allowed: zone_constraints.ride_start_allowed.unwrap_or(true), + ride_end_allowed: zone_constraints.ride_end_allowed.unwrap_or(true), + ride_through_allowed: zone_constraints.ride_through_allowed.unwrap_or(true), + maximum_speed_kph: zone_constraints.maximum_speed_kph, + station_parking: zone_constraints.station_parking.unwrap_or(true), + }; + Ok(result) + } +} + +fn to_datetime(value: Option) -> Result>, String> { + match value { + None => Ok(None), + Some(dt_string) => { + let dt = DateTime::parse_from_rfc3339(&dt_string) + .map_err(|e| format!("unable to parse date {dt_string}: {e}"))?; + Ok(Some(dt.to_utc())) + } + } +} diff --git a/rust/bambam-gbfs/src/model/gbfs/state.rs b/rust/bambam-gbfs/src/model/gbfs/state.rs new file mode 100644 index 00000000..65efff71 --- /dev/null +++ b/rust/bambam-gbfs/src/model/gbfs/state.rs @@ -0,0 +1,93 @@ +use std::collections::HashMap; + +use crate::model::gbfs::GbfsZoneRecord; + +/// aggregated record to represent the rules at a given time/place for use of GBFS. +/// built from the GBFS `geofencing_rules` for all zones that intersect with the +/// current destination of the graph search, flattened according to the rules of +/// precedence: . +pub struct ZoneState { + /// agency associated with this zone. + pub system_id: String, + /// Is the ride allowed to start in this zone? + pub ride_start_allowed: bool, + /// Is the ride allowed to end in this zone? + pub ride_end_allowed: bool, + /// Is the ride allowed to travel through this zone? + pub ride_through_allowed: bool, + /// What is the maximum speed allowed, in kilometers per hour? + pub maximum_speed_kph: Option, + /// Can vehicles only be parked at stations defined in [station_information] within this geofence zone? + pub station_parking: bool, +} + +impl ZoneState { + pub fn new(record: &GbfsZoneRecord) -> Self { + Self { + system_id: record.system_id.clone(), + ride_start_allowed: record.ride_start_allowed, + ride_end_allowed: record.ride_end_allowed, + ride_through_allowed: record.ride_through_allowed, + maximum_speed_kph: record.maximum_speed_kph, + station_parking: record.station_parking, + } + } + + // only overwrite values if they are the default when appending a lower-tier record, in accordance + // with the rules of precedence: . + pub fn append(&mut self, record: &GbfsZoneRecord) { + // Apply lowest index record values if an earlier record hasn't set it since we want highest precedence + if self.ride_start_allowed { + self.ride_start_allowed = record.ride_start_allowed; + } + if self.ride_end_allowed { + self.ride_end_allowed = record.ride_end_allowed; + } + if self.ride_through_allowed { + self.ride_through_allowed = record.ride_through_allowed; + } + if self.maximum_speed_kph.is_none() { + self.maximum_speed_kph = record.maximum_speed_kph; + } + if self.station_parking { + self.station_parking = record.station_parking; + } + } + + /// combines global and feature zone record "rules" as described in + /// . + /// + /// note: GbfsZoneRecords have already been "flattened" into concrete + /// boolean values by injecting the implicit default value "true" where + /// a field is omitted from the source rule. + /// + /// returns at most one zone per system_id. + pub fn collect_zones(intersection: Vec<&GbfsZoneRecord>) -> Vec { + let mut result: HashMap<&String, Self> = HashMap::new(); + + // Sort records by feature index so that the earliest overlapping polygon takes precedence. + let mut records: Vec<&GbfsZoneRecord> = intersection.clone(); + records.sort_by_key(|r| r.feature_index); + + // Iterate records from lowest index to highest, applying values and keeping the highest-precedence (lowest index). + for record in records { + result + .entry(&record.system_id) + .and_modify(|zone| zone.append(record)) + .or_insert(Self::new(record)); + } + + result.into_values().collect() + } + + /// used by sorting combinators so that the first value has the highest speed permissiveness + /// and, as a tie-breaker, the lexicagraphically-first system id. + /// + /// maximum_speed_kph values of None are treated as MAX values. + pub fn ascending_sort_key(&self) -> (i32, String) { + ( + -self.maximum_speed_kph.unwrap_or(i32::MAX), + self.system_id.clone(), + ) + } +} diff --git a/rust/bambam-gbfs/src/model/mod.rs b/rust/bambam-gbfs/src/model/mod.rs index e0a9a6f6..aec35c60 100644 --- a/rust/bambam-gbfs/src/model/mod.rs +++ b/rust/bambam-gbfs/src/model/mod.rs @@ -1,2 +1,4 @@ pub mod constraint; +pub mod feature; +pub mod gbfs; pub mod traversal; diff --git a/rust/bambam-gbfs/src/model/traversal/boarding/builder.rs b/rust/bambam-gbfs/src/model/traversal/boarding/builder.rs deleted file mode 100644 index 00d745e1..00000000 --- a/rust/bambam-gbfs/src/model/traversal/boarding/builder.rs +++ /dev/null @@ -1,23 +0,0 @@ -use std::sync::Arc; - -use routee_compass_core::model::traversal::{ - TraversalModelBuilder, TraversalModelError, TraversalModelService, -}; - -use super::{BoardingTraversalConfig, BoardingTraversalService}; - -pub struct BoardingTraversalBuilder {} - -impl TraversalModelBuilder for BoardingTraversalBuilder { - fn build( - &self, - parameters: &serde_json::Value, - ) -> Result, TraversalModelError> { - let config: BoardingTraversalConfig = serde_json::from_value(parameters.clone()) - .map_err(|e| TraversalModelError::BuildError(e.to_string()))?; - // this is where you will read GBFS files and store the data as fields - // on the GBFS traversal service. - let service = BoardingTraversalService::new(config); - Ok(Arc::new(service)) - } -} diff --git a/rust/bambam-gbfs/src/model/traversal/boarding/config.rs b/rust/bambam-gbfs/src/model/traversal/boarding/config.rs deleted file mode 100644 index 9cf0c0b4..00000000 --- a/rust/bambam-gbfs/src/model/traversal/boarding/config.rs +++ /dev/null @@ -1,8 +0,0 @@ -use serde::{Deserialize, Serialize}; - -/// this is where you will add the expected configuration fields required for -/// creating a GBFS service such as `gbfs_input_file: String` and any other -/// requirements for setting up GBFS for search. -#[derive(Serialize, Deserialize, Clone, Debug)] -#[serde(rename_all = "snake_case")] -pub struct BoardingTraversalConfig {} diff --git a/rust/bambam-gbfs/src/model/traversal/boarding/mod.rs b/rust/bambam-gbfs/src/model/traversal/boarding/mod.rs deleted file mode 100644 index 1f6bc6aa..00000000 --- a/rust/bambam-gbfs/src/model/traversal/boarding/mod.rs +++ /dev/null @@ -1,9 +0,0 @@ -mod builder; -mod config; -mod model; -mod service; - -pub use builder::BoardingTraversalBuilder; -pub use config::BoardingTraversalConfig; -pub use model::BoardingTraversalModel; -pub use service::BoardingTraversalService; diff --git a/rust/bambam-gbfs/src/model/traversal/boarding/model.rs b/rust/bambam-gbfs/src/model/traversal/boarding/model.rs deleted file mode 100644 index 84ad565a..00000000 --- a/rust/bambam-gbfs/src/model/traversal/boarding/model.rs +++ /dev/null @@ -1,45 +0,0 @@ -use routee_compass_core::{ - algorithm::search::SearchTree, - model::{ - network::Vertex, - state::{InputFeature, StateModel, StateVariable, StateVariableConfig}, - traversal::{EdgeFrontierContext, TraversalModel, TraversalModelError}, - }, -}; - -/// applies wait times when boarding a micromobility vehicle. -pub struct BoardingTraversalModel {} - -impl TraversalModel for BoardingTraversalModel { - fn name(&self) -> String { - "BoardingTraversalModel".to_string() - } - - fn input_features(&self) -> Vec { - todo!() - } - - fn output_features(&self) -> Vec<(String, StateVariableConfig)> { - todo!() - } - - fn estimate_traversal( - &self, - _od: (&Vertex, &Vertex), - _state: &mut Vec, - _tree: &SearchTree, - _state_model: &StateModel, - ) -> Result<(), TraversalModelError> { - // this can be skipped if we aren't trying to use A*. - Ok(()) - } - - fn traverse_edge( - &self, - _ctx: &EdgeFrontierContext, - _state: &mut Vec, - _state_model: &StateModel, - ) -> Result<(), TraversalModelError> { - todo!() - } -} diff --git a/rust/bambam-gbfs/src/model/traversal/boarding/service.rs b/rust/bambam-gbfs/src/model/traversal/boarding/service.rs deleted file mode 100644 index dfde5858..00000000 --- a/rust/bambam-gbfs/src/model/traversal/boarding/service.rs +++ /dev/null @@ -1,28 +0,0 @@ -use std::sync::Arc; - -use routee_compass_core::model::traversal::{ - TraversalModel, TraversalModelError, TraversalModelService, -}; - -use super::BoardingTraversalConfig; - -pub struct BoardingTraversalService { - pub config: BoardingTraversalConfig, -} - -impl BoardingTraversalService { - pub fn new(config: BoardingTraversalConfig) -> BoardingTraversalService { - BoardingTraversalService { config } - } -} - -impl TraversalModelService for BoardingTraversalService { - fn build( - &self, - _query: &serde_json::Value, - ) -> Result, TraversalModelError> { - // if there's anything that can change between the execution of each search, - // we should attempt to pull it from the query here. - todo!() - } -} diff --git a/rust/bambam-gbfs/src/model/traversal/gbfs_traversal/builder.rs b/rust/bambam-gbfs/src/model/traversal/gbfs_traversal/builder.rs new file mode 100644 index 00000000..add48801 --- /dev/null +++ b/rust/bambam-gbfs/src/model/traversal/gbfs_traversal/builder.rs @@ -0,0 +1,27 @@ +use std::sync::Arc; + +use super::{GbfsTraversalConfig, GbfsTraversalEngine, GbfsTraversalService}; + +use routee_compass_core::model::traversal::{ + TraversalModelBuilder, TraversalModelError, TraversalModelService, +}; + +pub struct GbfsTraversalBuilder {} + +impl TraversalModelBuilder for GbfsTraversalBuilder { + fn build( + &self, + value: &serde_json::Value, + ) -> Result, TraversalModelError> { + let config: GbfsTraversalConfig = serde_json::from_value(value.clone()).map_err(|e| { + let msg = format!("failure reading config for GbfsTraversal builder: {e}"); + TraversalModelError::BuildError(msg) + })?; + let engine = GbfsTraversalEngine::try_from(config).map_err(|e| { + let msg = format!("failure building engine from config for GbfsTraversal builder: {e}"); + TraversalModelError::BuildError(msg) + })?; + let service = GbfsTraversalService::new(engine); + Ok(Arc::new(service)) + } +} diff --git a/rust/bambam-gbfs/src/model/traversal/gbfs_traversal/config.rs b/rust/bambam-gbfs/src/model/traversal/gbfs_traversal/config.rs new file mode 100644 index 00000000..856de3be --- /dev/null +++ b/rust/bambam-gbfs/src/model/traversal/gbfs_traversal/config.rs @@ -0,0 +1,25 @@ +use routee_compass_core::model::unit::SpeedUnit; +use serde::{Deserialize, Serialize}; + +#[derive(Deserialize, Serialize, Clone, Debug)] +pub struct GbfsTraversalConfig { + /// output of bambam-gbfs CLI import process, contains a record of the + /// identifier, optional start/end times for service, and traversal ruleset + /// for default vehicles (trips without a VehicleTripId). + /// + /// see [crate::model::gbfs::GbfsZoneRecord] + pub zone_record_input_file: String, + /// output of bambam-gbfs CLI import process, contains zonal geometries + /// with matching indices to the zones input file. + pub zone_geometry_input_file: String, + /// system identifiers for all records. these are sorted lexicagraphically. + pub system_ids_input_file: String, + /// speed to use for GBFS trips. can be limited by zone-specific max speeds. + pub default_speed: DefaultSpeed, +} + +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct DefaultSpeed { + pub speed: f64, + pub speed_unit: SpeedUnit, +} diff --git a/rust/bambam-gbfs/src/model/traversal/gbfs_traversal/engine.rs b/rust/bambam-gbfs/src/model/traversal/gbfs_traversal/engine.rs new file mode 100644 index 00000000..6a7c92f1 --- /dev/null +++ b/rust/bambam-gbfs/src/model/traversal/gbfs_traversal/engine.rs @@ -0,0 +1,174 @@ +use std::path::Path; + +use crate::model::{ + feature, + gbfs::{GbfsLookupModel, ZoneState}, +}; + +use super::GbfsTraversalConfig; + +use bambam_core::model::state::CategoricalStateMapping; +use chrono::{DateTime, Utc}; +use itertools::Itertools; +use routee_compass_core::model::{ + constraint::ConstraintModelError, + network::Vertex, + state::{StateModel, StateVariable}, + traversal::TraversalModelError, +}; + +pub struct GbfsTraversalEngine { + lookup: GbfsLookupModel, + mapping: CategoricalStateMapping, + pub default_speed: uom::si::f64::Velocity, +} + +impl GbfsTraversalEngine { + /// runs the traversal logic for the GBFS traversal model at an iteration of graph search. + pub fn traverse( + &self, + vertex: &Vertex, + state: &mut [StateVariable], + state_model: &StateModel, + start_time: DateTime, + ) -> Result<(), TraversalModelError> { + // are we currently boarded with a system_id? + let service_opt = feature::state::get_system_id(state, state_model, &self.mapping) + .map_err(|e| { + let msg = format!("failure inspecting gbfs service id of search state: {e}"); + TraversalModelError::TraversalModelFailure(msg) + })?; + + // find intersecting zones. if we are already boarded, only accept zones with our existing system_id. + let zones = self + .lookup + .matching_zones(vertex, state, state_model, start_time, service_opt) + .map_err(|e| { + let vertex_id = vertex.vertex_id; + let service_msg = match service_opt { + Some(id) => format!("for trip boarded on service '{id}'"), + None => "for unboarded trip".to_string(), + }; + let msg = format!( + "failure getting gbfs zone rules for vertex {vertex_id} {service_msg}: {e}" + ); + TraversalModelError::TraversalModelFailure(msg) + })?; + + match service_opt { + Some(_) => process_boarded(&zones, self.default_speed, state, state_model), + None => process_unboarded(zones, self.default_speed, state, state_model, &self.mapping), + } + } +} + +impl TryFrom for GbfsTraversalEngine { + type Error = ConstraintModelError; + + fn try_from(config: GbfsTraversalConfig) -> Result { + let lookup = GbfsLookupModel::new( + &config.zone_record_input_file, + &config.zone_geometry_input_file, + ) + .map_err(|e| { + let msg = format!("failure building GBFS lookup model: {e}"); + ConstraintModelError::BuildError(msg) + })?; + + let mapping = CategoricalStateMapping::from_enumerated_category_file(Path::new( + &config.system_ids_input_file, + )) + .map_err(|e| { + ConstraintModelError::BuildError(format!( + "failure while building categorical mapping from {}: {e}", + config.system_ids_input_file + )) + })?; + + let default_speed = config + .default_speed + .speed_unit + .to_uom(config.default_speed.speed); + Ok(Self { + lookup, + mapping, + default_speed, + }) + } +} + +/// traversal is associated with a trip that has already boarded GBFS. here we must +// 1. determine if this location is a valid destination, and set if true (ride_end_allowed) +// 2. write edge_speed (either from config value or overriding value from record) +// 3. (todo) if station_parking -> confirm we are at a station location +fn process_boarded( + zones: &[ZoneState], + default_speed: uom::si::f64::Velocity, + state: &mut [StateVariable], + state_model: &StateModel, +) -> Result<(), TraversalModelError> { + use uom::si::{f64::Velocity, velocity::kilometer_per_hour}; + let zone = match zones { + [zone] => zone, + _ => { + let n_zones = zones.len(); + let msg = format!( + "expected a boarded GBFS trip would match exactly one zone state, found {n_zones}." + ); + return Err(TraversalModelError::TraversalModelFailure(msg)); + } + }; + + // 1. is this a valid destination? by default, we assume yes, but here + // we overwrite as false if `ride_end_allowed` is false. + if !zone.ride_end_allowed { + feature::state::set_invalid_destination(state, state_model)?; + } + // 2. set the speed, limiting if max speed is present + let speed_value: Velocity = match zone.maximum_speed_kph { + Some(max) => { + let max_vel = Velocity::new::(max.into()); + max_vel.min(default_speed) + } + None => default_speed, + }; + state_model.set_speed(state, "edge_speed", &speed_value)?; + + Ok(()) +} + +/// traversal is associated with a trip that has not yet boarded GBFS. here we must +/// 1. pick the best-quality system from the intersecting zones +/// - ride_start_allowed is true +/// - best speed +/// - sort ids lexicagraphically +/// 2. board with that system_id +/// 3. process boarded +fn process_unboarded( + zones: Vec, + default_speed: uom::si::f64::Velocity, + state: &mut [StateVariable], + state_model: &StateModel, + mapping: &CategoricalStateMapping, +) -> Result<(), TraversalModelError> { + let n_zones = zones.len(); + + let best_zone = zones + .into_iter() + .filter(|z| z.ride_start_allowed && z.ride_through_allowed) + .sorted_by_cached_key(|z| z.ascending_sort_key()) + .next(); + + match best_zone { + Some(best) => { + feature::state::set_system_id(state, state_model, &best.system_id, mapping)?; + process_boarded(&[best], default_speed, state, state_model) + } + None => { + let msg = format!( + "found {n_zones} zones but none are ride_start_allowed; should have been caught by constraint model!", + ); + Err(TraversalModelError::InternalError(msg)) + } + } +} diff --git a/rust/bambam-gbfs/src/model/traversal/gbfs_traversal/mod.rs b/rust/bambam-gbfs/src/model/traversal/gbfs_traversal/mod.rs new file mode 100644 index 00000000..ad94cf12 --- /dev/null +++ b/rust/bambam-gbfs/src/model/traversal/gbfs_traversal/mod.rs @@ -0,0 +1,13 @@ +mod builder; +mod config; +mod engine; +mod model; +mod params; +mod service; + +pub use builder::GbfsTraversalBuilder; +pub use config::GbfsTraversalConfig; +pub use engine::GbfsTraversalEngine; +pub use model::GbfsTraversalModel; +pub use params::GbfsTraversalParams; +pub use service::GbfsTraversalService; diff --git a/rust/bambam-gbfs/src/model/traversal/gbfs_traversal/model.rs b/rust/bambam-gbfs/src/model/traversal/gbfs_traversal/model.rs new file mode 100644 index 00000000..c50b7a40 --- /dev/null +++ b/rust/bambam-gbfs/src/model/traversal/gbfs_traversal/model.rs @@ -0,0 +1,82 @@ +use std::sync::Arc; + +use crate::model::feature; + +use super::{GbfsTraversalEngine, GbfsTraversalParams}; + +use routee_compass_core::{ + algorithm::search::SearchTree, + model::{ + network::Vertex, + state::{InputFeature, StateModel, StateVariable, StateVariableConfig}, + traversal::{EdgeFrontierContext, TraversalModel, TraversalModelError}, + }, +}; +use uom::si::f64::Velocity; + +pub struct GbfsTraversalModel { + pub engine: Arc, + pub params: GbfsTraversalParams, +} + +impl GbfsTraversalModel { + pub fn new(engine: Arc, params: GbfsTraversalParams) -> Self { + // modify this and the struct definition if additional pre-processing + // is required during model instantiation from query parameters. + Self { engine, params } + } +} + +impl TraversalModel for GbfsTraversalModel { + fn name(&self) -> String { + "GbfsTraversalModel".to_string() + } + + fn input_features(&self) -> Vec { + vec![] + } + + fn output_features(&self) -> Vec<(String, StateVariableConfig)> { + // 1. valid destination + // 2. max_speed + edge_speed + vec![ + ( + feature::fieldname::GBFS_DESTINATION.to_string(), + feature::variable::gbfs_destination(), + ), + ( + feature::fieldname::GBFS_SYSTEM_ID.to_string(), + feature::variable::gbfs_system_id(), + ), + ( + "edge_speed".to_string(), + StateVariableConfig::Speed { + initial: Velocity::new::(0.0), + accumulator: false, + output_unit: None, + }, + ), + ] + } + + fn traverse_edge( + &self, + ctx: &EdgeFrontierContext, + state: &mut Vec, + state_model: &StateModel, + ) -> Result<(), TraversalModelError> { + self.engine + .traverse(ctx.dst, state, state_model, self.params.start_time) + } + + fn estimate_traversal( + &self, + _od: (&Vertex, &Vertex), + state: &mut Vec, + _tree: &SearchTree, + state_model: &StateModel, + ) -> Result<(), TraversalModelError> { + state_model.set_speed(state, "edge_speed", &self.engine.default_speed)?; + Ok(()) + } +} diff --git a/rust/bambam-gbfs/src/model/traversal/gbfs_traversal/params.rs b/rust/bambam-gbfs/src/model/traversal/gbfs_traversal/params.rs new file mode 100644 index 00000000..8c78c9fb --- /dev/null +++ b/rust/bambam-gbfs/src/model/traversal/gbfs_traversal/params.rs @@ -0,0 +1,8 @@ +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; + +#[derive(Deserialize, Serialize, Clone, Debug)] +pub struct GbfsTraversalParams { + /// datetime we start this trip. + pub start_time: DateTime, +} diff --git a/rust/bambam-gbfs/src/model/traversal/gbfs_traversal/service.rs b/rust/bambam-gbfs/src/model/traversal/gbfs_traversal/service.rs new file mode 100644 index 00000000..a2b4f050 --- /dev/null +++ b/rust/bambam-gbfs/src/model/traversal/gbfs_traversal/service.rs @@ -0,0 +1,33 @@ +use std::sync::Arc; + +use super::{GbfsTraversalEngine, GbfsTraversalModel, GbfsTraversalParams}; + +use routee_compass_core::model::traversal::{ + TraversalModel, TraversalModelError, TraversalModelService, +}; + +pub struct GbfsTraversalService { + engine: Arc, +} + +impl GbfsTraversalService { + pub fn new(engine: GbfsTraversalEngine) -> Self { + Self { + engine: Arc::new(engine), + } + } +} + +impl TraversalModelService for GbfsTraversalService { + fn build( + &self, + query: &serde_json::Value, + ) -> Result, TraversalModelError> { + let params: GbfsTraversalParams = serde_json::from_value(query.clone()).map_err(|e| { + let msg = format!("failure reading params for GbfsTraversal service: {e}"); + TraversalModelError::BuildError(msg) + })?; + let model = GbfsTraversalModel::new(self.engine.clone(), params); + Ok(Arc::new(model)) + } +} diff --git a/rust/bambam-gbfs/src/model/traversal/mod.rs b/rust/bambam-gbfs/src/model/traversal/mod.rs index dd807c51..55b32aa1 100644 --- a/rust/bambam-gbfs/src/model/traversal/mod.rs +++ b/rust/bambam-gbfs/src/model/traversal/mod.rs @@ -1 +1 @@ -pub mod boarding; +pub mod gbfs_traversal; diff --git a/rust/bambam-osm/src/model/osm/graph/compass_writer.rs b/rust/bambam-osm/src/model/osm/graph/compass_writer.rs index 994114ff..e10566cb 100644 --- a/rust/bambam-osm/src/model/osm/graph/compass_writer.rs +++ b/rust/bambam-osm/src/model/osm/graph/compass_writer.rs @@ -39,7 +39,7 @@ impl CompassWriter for OsmGraphVectorized { let dirname = output_directory.as_os_str().to_string_lossy(); return Err(OsmError::InternalError(format!( "unable to create directory {}", - &dirname + dirname ))); } diff --git a/rust/bambam-py/Cargo.toml b/rust/bambam-py/Cargo.toml index d849f589..6d00bd4e 100644 --- a/rust/bambam-py/Cargo.toml +++ b/rust/bambam-py/Cargo.toml @@ -14,7 +14,7 @@ inventory = { workspace = true } itertools = { workspace = true } -pyo3 = { version = "0.29.0", features = ["extension-module", "serde"] } +pyo3 = { workspace = true } routee-compass = { workspace = true, default-features = false } routee-compass-core = { workspace = true } routee-compass-macros = { workspace = true } diff --git a/rust/bambam/src/model/builders.rs b/rust/bambam/src/model/builders.rs index 35e5d15d..7b2a7485 100644 --- a/rust/bambam/src/model/builders.rs +++ b/rust/bambam/src/model/builders.rs @@ -11,9 +11,8 @@ use crate::model::output_plugin::isochrone::isochrone_output_plugin_builder::Iso use crate::model::output_plugin::opportunity::OpportunityOutputPluginBuilder; use crate::model::traversal::multimodal::MultimodalTraversalBuilder; use crate::model::traversal::switch::switch_traversal_builder::SwitchTraversalBuilder; -use bambam_gbfs::model::constraint::boarding::BoardingConstraintBuilder; -use bambam_gbfs::model::constraint::geofence::GeofenceConstraintBuilder; -use bambam_gbfs::model::traversal::boarding::BoardingTraversalBuilder; +use bambam_gbfs::model::constraint::gbfs_constraint::GbfsConstraintBuilder; +use bambam_gbfs::model::traversal::gbfs_traversal::GbfsTraversalBuilder; use bambam_gtfs::model::traversal::transit::TransitTraversalBuilder; use bambam_gtfs_flex::model::constraint::GtfsFlexDepartureFrontierBuilder; use bambam_gtfs_flex::model::traversal::flex::GtfsFlexBuilder; @@ -41,18 +40,8 @@ pub const BUILDER_REGISTRATION: BuilderRegistration = BuilderRegistration(|build ); builders.add_traversal_model(String::from("transit"), Rc::new(TransitTraversalBuilder {})); - builders.add_constraint_model( - "gbfs_geofence".to_string(), - Rc::new(GeofenceConstraintBuilder {}), - ); - builders.add_constraint_model( - "gbfs_boarding".to_string(), - Rc::new(BoardingConstraintBuilder {}), - ); - builders.add_traversal_model( - "gbfs_boarding".to_string(), - Rc::new(BoardingTraversalBuilder {}), - ); + builders.add_constraint_model("gbfs".to_string(), Rc::new(GbfsConstraintBuilder {})); + builders.add_traversal_model("gbfs".to_string(), Rc::new(GbfsTraversalBuilder {})); builders.add_traversal_model("gtfs-flex".to_string(), Rc::new(GtfsFlexBuilder {})); diff --git a/rust/bambam/src/model/input_plugin/population/population_source_config.rs b/rust/bambam/src/model/input_plugin/population/population_source_config.rs index f22f9640..6a3acda1 100644 --- a/rust/bambam/src/model/input_plugin/population/population_source_config.rs +++ b/rust/bambam/src/model/input_plugin/population/population_source_config.rs @@ -20,7 +20,10 @@ pub enum PopulationSourceConfig { } fn env_census_api_token() -> Result { - std::env::var("CENSUS_API_TOKEN").map_err(|e| format!("ACS token required, {e}")) + std::env::var("CENSUS_API_TOKEN").map_err(|_| { + "ACS requires token for access. please set via 'CENSUS_API_TOKEN' environment variable" + .to_string() + }) } impl PopulationSourceConfig { diff --git a/rust/bambam/src/model/traversal/time_delay/time_delay_record.rs b/rust/bambam/src/model/traversal/time_delay/time_delay_record.rs index edc7adef..2023dac0 100644 --- a/rust/bambam/src/model/traversal/time_delay/time_delay_record.rs +++ b/rust/bambam/src/model/traversal/time_delay/time_delay_record.rs @@ -59,7 +59,7 @@ impl<'de> de::Deserialize<'de> for TimeDelayRecord { .map_err(|e| { de::Error::custom(format!( "unable to parse WKT geometry '{}': {}", - &value, e + value, e )) })?; let row_geometry: Geometry = match geo_f64 { @@ -78,7 +78,7 @@ impl<'de> de::Deserialize<'de> for TimeDelayRecord { _ => { return Err(de::Error::custom(format!( "expected Polygon or MultiPolygon geometry, found unexpected type in '{}'", - &value + value ))); } }; @@ -89,7 +89,7 @@ impl<'de> de::Deserialize<'de> for TimeDelayRecord { serde_json::from_str::