85 lines
3.3 KiB
Rust
85 lines
3.3 KiB
Rust
use log::debug;
|
|
/***
|
|
* Implementation of GTFS-R polling
|
|
*/
|
|
use reqwest::{StatusCode, blocking::Client};
|
|
use std::{collections::HashMap, time::Duration};
|
|
|
|
use crate::gtfs::{self, structs::{Arrival, Error, Gtfs}};
|
|
|
|
impl Gtfs<'_> {
|
|
pub(crate) fn realtime_update (&self, arrivals: &mut Vec<Arrival<'_>>) -> Result<(), gtfs::structs::Error>{
|
|
|
|
// Poll GTFS-R API
|
|
let client = Client::builder()
|
|
.timeout(Duration::from_secs(10))
|
|
.connect_timeout(Duration::from_secs(5))
|
|
.build()?;
|
|
|
|
let response = client
|
|
.get(&self.preferences.realtime_url)
|
|
.header("x-api-key", &self.preferences.realtime_api_key)
|
|
.send()?;
|
|
|
|
if response.status() != StatusCode::OK {
|
|
return Err(Error {
|
|
_message : format!("HTTP Respomse: {:#?}. Payload: \n{:#?}\n", response.status(), response.text().unwrap())
|
|
})
|
|
}
|
|
|
|
// Parse response
|
|
let response_bytes= response.bytes()?;
|
|
let data: Result<gtfs_realtime::FeedMessage, prost::DecodeError> = prost::Message::decode(response_bytes.as_ref());
|
|
if data.is_err() {
|
|
return Err(Error {
|
|
_message : format! ("Error loading realtime prtobuf: {:#?}", data.err().unwrap())
|
|
})
|
|
}
|
|
|
|
|
|
// Build a map of (trip, stop) -> Arrival for faster lookup
|
|
let mut lookup: HashMap<(String, String), usize> = HashMap::new();
|
|
let num_arrivals = arrivals.len();
|
|
for i in 0..num_arrivals {
|
|
let arrival = arrivals.get(i).unwrap();
|
|
lookup.insert((arrival.trip.id.clone(), arrival.stop.id.clone()), i);
|
|
}
|
|
|
|
// Match deltas to existing arrivals
|
|
let entities = data.unwrap().entity;
|
|
debug!("Loaded {} entities from realtime update.", entities.len());
|
|
for entity in entities {
|
|
if entity.trip_update.is_some() && entity.stop.is_some() {
|
|
let trip_update = entity.trip_update.unwrap();
|
|
|
|
// Look up the entry in arrivals for the current
|
|
for stop_update in trip_update.stop_time_update {
|
|
if stop_update.stop_id.is_none() {
|
|
continue;
|
|
}
|
|
let trip_id = trip_update.trip.trip_id.clone().unwrap();
|
|
let stop_id = stop_update.stop_id.unwrap();
|
|
let key = (trip_id, stop_id);
|
|
if lookup.contains_key(&key) {
|
|
|
|
// Update the arrival time
|
|
let arrival_index = lookup.get(&key);
|
|
let arrival: &mut Arrival<'_> = arrivals.get_mut(*arrival_index.unwrap()).unwrap();
|
|
if let Some(update) = stop_update.departure.or(stop_update.arrival) {
|
|
let new_time = (((arrival.departure_time as i64)
|
|
+ (update.delay.unwrap_or(0) as i64))) as u32;
|
|
debug!("Updated arrival time of {:#?} by {}s", &key, update.delay.unwrap());
|
|
arrival.departure_time = new_time;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
return Ok(());
|
|
}
|
|
}
|
|
|
|
|
|
|