1use serde::Deserialize;
2use sha2::{Digest, Sha256};
3
4use crate::{Error, Gtfs, RawGtfs};
5use std::collections::HashMap;
6use std::convert::TryFrom;
7use std::fs::File;
8use std::io::Read;
9use std::path::Path;
10use web_time::Instant;
11
12pub struct GtfsReader {
35 pub read_stop_times: bool,
37 pub read_shapes: bool,
39 pub unkown_enum_as_default: bool,
41 pub trim_fields: bool,
46}
47
48impl Default for GtfsReader {
49 fn default() -> Self {
50 GtfsReader {
51 read_stop_times: true,
52 read_shapes: true,
53 unkown_enum_as_default: false,
54 trim_fields: true,
55 }
56 }
57}
58
59impl GtfsReader {
60 pub fn read_stop_times(mut self, read_stop_times: bool) -> Self {
65 self.read_stop_times = read_stop_times;
66 self
67 }
68
69 pub fn read_shapes(mut self, read_shapes: bool) -> Self {
72 self.read_shapes = read_shapes;
73 self
74 }
75
76 pub fn unkown_enum_as_default(mut self, unkown_enum_as_default: bool) -> Self {
83 self.unkown_enum_as_default = unkown_enum_as_default;
84 self
85 }
86
87 pub fn trim_fields(mut self, trim_fields: bool) -> Self {
92 self.trim_fields = trim_fields;
93 self
94 }
95
96 #[cfg(not(target_arch = "wasm32"))]
101 pub fn read(self, gtfs: &str) -> Result<Gtfs, Error> {
102 self.raw().read(gtfs).and_then(Gtfs::try_from)
103 }
104
105 pub fn read_from_path<P>(self, path: P) -> Result<Gtfs, Error>
107 where
108 P: AsRef<Path>,
109 {
110 self.raw().read_from_path(path).and_then(Gtfs::try_from)
111 }
112
113 #[cfg(all(feature = "read-url", not(target_arch = "wasm32")))]
117 pub fn read_from_url<U: reqwest::IntoUrl>(self, url: U) -> Result<Gtfs, Error> {
118 self.raw().read_from_url(url).and_then(Gtfs::try_from)
119 }
120
121 #[cfg(feature = "read-url")]
125 pub async fn read_from_url_async<U: reqwest::IntoUrl>(self, url: U) -> Result<Gtfs, Error> {
126 self.raw()
127 .read_from_url_async(url)
128 .await
129 .and_then(Gtfs::try_from)
130 }
131
132 pub fn raw(self) -> RawGtfsReader {
144 RawGtfsReader { reader: self }
145 }
146}
147
148pub struct RawGtfsReader {
152 reader: GtfsReader,
153}
154
155impl RawGtfsReader {
156 fn read_from_directory(&self, p: &std::path::Path) -> Result<RawGtfs, Error> {
157 let start_of_read_instant = Instant::now();
158 let files = std::fs::read_dir(p)?
161 .filter_map(|d| {
162 d.ok().and_then(|e| {
163 e.path()
164 .strip_prefix(p)
165 .ok()
166 .and_then(|f| f.to_str().map(|s| s.to_owned()))
167 })
168 })
169 .collect();
170
171 let mut result = RawGtfs {
172 trips: self.read_objs_from_path(p.join("trips.txt")),
173 calendar: self.read_objs_from_optional_path(p, "calendar.txt"),
174 calendar_dates: self.read_objs_from_optional_path(p, "calendar_dates.txt"),
175 stops: self.read_objs_from_path(p.join("stops.txt")),
176 routes: self.read_objs_from_path(p.join("routes.txt")),
177 stop_times: if self.reader.read_stop_times {
178 self.read_objs_from_path(p.join("stop_times.txt"))
179 } else {
180 Ok(Vec::new())
181 },
182 agencies: self.read_objs_from_path(p.join("agency.txt")),
183 shapes: self.read_objs_from_optional_path(p, "shapes.txt"),
184 fare_attributes: self.read_objs_from_optional_path(p, "fare_attributes.txt"),
185 fare_rules: self.read_objs_from_optional_path(p, "fare_rules.txt"),
186 fare_products: self.read_objs_from_optional_path(p, "fare_products.txt"),
187 fare_media: self.read_objs_from_optional_path(p, "fare_media.txt"),
188 rider_categories: self.read_objs_from_optional_path(p, "rider_categories.txt"),
189 frequencies: self.read_objs_from_optional_path(p, "frequencies.txt"),
190 transfers: self.read_objs_from_optional_path(p, "transfers.txt"),
191 pathways: self.read_objs_from_optional_path(p, "pathways.txt"),
192 feed_info: self.read_objs_from_optional_path(p, "feed_info.txt"),
193 read_duration: start_of_read_instant.elapsed(),
194 translations: self.read_objs_from_optional_path(p, "translations.txt"),
195 ticketing_deep_links: self.read_objs_from_optional_path(p, "ticketing_deep_links.txt"),
196 ticketing_identifiers: self
197 .read_objs_from_optional_path(p, "ticketing_identifiers.txt"),
198 attributions: self.read_objs_from_optional_path(p, "attributions.txt"),
199 files,
200 source_format: crate::SourceFormat::Directory,
201 sha256: None,
202 };
203
204 if self.reader.unkown_enum_as_default {
205 result.unknown_to_default();
206 }
207 Ok(result)
208 }
209
210 #[cfg(not(target_arch = "wasm32"))]
213 pub fn read(self, gtfs: &str) -> Result<RawGtfs, Error> {
214 #[cfg(feature = "read-url")]
215 if gtfs.starts_with("http") {
216 return self.read_from_url(gtfs);
217 }
218 self.read_from_path(gtfs)
219 }
220
221 #[cfg(all(feature = "read-url", not(target_arch = "wasm32")))]
223 pub fn read_from_url<U: reqwest::IntoUrl>(self, url: U) -> Result<RawGtfs, Error> {
224 let mut res = reqwest::blocking::get(url)?;
225 let mut body = Vec::new();
226 res.read_to_end(&mut body)?;
227 let cursor = std::io::Cursor::new(body);
228 self.read_from_reader(cursor)
229 }
230
231 #[cfg(feature = "read-url")]
233 pub async fn read_from_url_async<U: reqwest::IntoUrl>(self, url: U) -> Result<RawGtfs, Error> {
234 let res = reqwest::get(url).await?.bytes().await?;
235 let reader = std::io::Cursor::new(res);
236 self.read_from_reader(reader)
237 }
238
239 pub fn read_from_path<P>(&self, path: P) -> Result<RawGtfs, Error>
241 where
242 P: AsRef<Path>,
243 {
244 let p = path.as_ref();
245 if p.is_file() {
246 let reader = File::open(p)?;
247 self.read_from_reader(reader)
248 } else if p.is_dir() {
249 self.read_from_directory(p)
250 } else {
251 Err(Error::NotFileNorDirectory(format!("{}", p.display())))
252 }
253 }
254
255 pub fn read_from_reader<T: std::io::Read + std::io::Seek>(
256 &self,
257 reader: T,
258 ) -> Result<RawGtfs, Error> {
259 let start_of_read_instant = Instant::now();
260 let hasher = Sha256::new();
261 let mut buf_reader = std::io::BufReader::new(reader);
262 let mut hash_io = digest_io::IoWrapper(hasher);
263 let _n = std::io::copy(&mut buf_reader, &mut hash_io)?;
264 let digest_io::IoWrapper(hasher) = hash_io;
265 let hash = hasher.finalize();
266 let mut archive = zip::ZipArchive::new(buf_reader)?;
267 let mut file_mapping = HashMap::new();
268 let mut files = Vec::new();
269
270 for i in 0..archive.len() {
271 let archive_file = archive.by_index(i)?;
272 files.push(archive_file.name().to_owned());
273
274 for gtfs_file in &[
275 "agency.txt",
276 "calendar.txt",
277 "calendar_dates.txt",
278 "routes.txt",
279 "stops.txt",
280 "stop_times.txt",
281 "trips.txt",
282 "fare_attributes.txt",
283 "fare_rules.txt",
284 "fare_products.txt",
285 "fare_media.txt",
286 "rider_categories.txt",
287 "frequencies.txt",
288 "transfers.txt",
289 "pathways.txt",
290 "feed_info.txt",
291 "shapes.txt",
292 "translations.txt",
293 "ticketing_deep_links.txt",
294 "ticketing_identifiers.txt",
295 "attributions.txt",
296 ] {
297 let path = std::path::Path::new(archive_file.name());
298 if path.file_name() == Some(std::ffi::OsStr::new(gtfs_file)) {
299 file_mapping.insert(gtfs_file, i);
300 break;
301 }
302 }
303 }
304
305 let mut result = RawGtfs {
306 agencies: self.read_file(&file_mapping, &mut archive, "agency.txt"),
307 calendar: self.read_optional_file(&file_mapping, &mut archive, "calendar.txt"),
308 calendar_dates: self.read_optional_file(
309 &file_mapping,
310 &mut archive,
311 "calendar_dates.txt",
312 ),
313 routes: self.read_file(&file_mapping, &mut archive, "routes.txt"),
314 stops: self.read_file(&file_mapping, &mut archive, "stops.txt"),
315 stop_times: if self.reader.read_stop_times {
316 self.read_file(&file_mapping, &mut archive, "stop_times.txt")
317 } else {
318 Ok(Vec::new())
319 },
320 trips: self.read_file(&file_mapping, &mut archive, "trips.txt"),
321 fare_attributes: self.read_optional_file(
322 &file_mapping,
323 &mut archive,
324 "fare_attributes.txt",
325 ),
326 fare_rules: self.read_optional_file(&file_mapping, &mut archive, "fare_rules.txt"),
327 fare_products: self.read_optional_file(
328 &file_mapping,
329 &mut archive,
330 "fare_products.txt",
331 ),
332 fare_media: self.read_optional_file(&file_mapping, &mut archive, "fare_media.txt"),
333 rider_categories: self.read_optional_file(
334 &file_mapping,
335 &mut archive,
336 "rider_categories.txt",
337 ),
338 frequencies: self.read_optional_file(&file_mapping, &mut archive, "frequencies.txt"),
339 transfers: self.read_optional_file(&file_mapping, &mut archive, "transfers.txt"),
340 pathways: self.read_optional_file(&file_mapping, &mut archive, "pathways.txt"),
341 feed_info: self.read_optional_file(&file_mapping, &mut archive, "feed_info.txt"),
342 shapes: if self.reader.read_shapes {
343 self.read_optional_file(&file_mapping, &mut archive, "shapes.txt")
344 } else {
345 Some(Ok(Vec::new()))
346 },
347 translations: self.read_optional_file(&file_mapping, &mut archive, "translations.txt"),
348 ticketing_deep_links: self.read_optional_file(
349 &file_mapping,
350 &mut archive,
351 "ticketing_deep_links.txt",
352 ),
353 ticketing_identifiers: self.read_optional_file(
354 &file_mapping,
355 &mut archive,
356 "ticketing_identifiers.txt",
357 ),
358 attributions: self.read_optional_file(&file_mapping, &mut archive, "attributions.txt"),
359 read_duration: start_of_read_instant.elapsed(),
360 files,
361 source_format: crate::SourceFormat::Zip,
362 sha256: Some(base16ct::lower::encode_string(&hash)),
363 };
364
365 if self.reader.unkown_enum_as_default {
366 result.unknown_to_default();
367 }
368 Ok(result)
369 }
370
371 fn read_objs<T, O>(&self, mut reader: T, file_name: &str) -> Result<Vec<O>, Error>
372 where
373 for<'de> O: Deserialize<'de>,
374 T: std::io::Read,
375 {
376 let mut bom = [0; 3];
377 reader
378 .read_exact(&mut bom)
379 .map_err(|e| Error::NamedFileIO {
380 file_name: file_name.to_owned(),
381 source: Box::new(e),
382 })?;
383
384 let chained = if bom != [0xefu8, 0xbbu8, 0xbfu8] {
385 bom.chain(reader)
386 } else {
387 [].chain(reader)
388 };
389
390 let mut reader = csv::ReaderBuilder::new()
391 .flexible(true)
392 .trim(if self.reader.trim_fields {
393 csv::Trim::Fields
394 } else {
395 csv::Trim::None
396 })
397 .from_reader(chained);
398 let headers = reader
400 .headers()
401 .map_err(|e| Error::CSVError {
402 file_name: file_name.to_owned(),
403 source: e,
404 line_in_error: None,
405 })?
406 .clone()
407 .into_iter()
408 .map(|x| x.trim())
409 .collect::<csv::StringRecord>();
410
411 let mut rec = csv::StringRecord::new();
413 let mut objs = Vec::new();
414
415 while reader.read_record(&mut rec).map_err(|e| Error::CSVError {
417 file_name: file_name.to_owned(),
418 source: e,
419 line_in_error: None,
420 })? {
421 let obj = rec
422 .deserialize(Some(&headers))
423 .map_err(|e| Error::CSVError {
424 file_name: file_name.to_owned(),
425 source: e,
426 line_in_error: Some(crate::error::LineError {
427 headers: headers.into_iter().map(String::from).collect(),
428 values: rec.into_iter().map(String::from).collect(),
429 }),
430 })?;
431 objs.push(obj);
432 }
433 Ok(objs)
434 }
435
436 fn read_objs_from_path<O>(&self, path: std::path::PathBuf) -> Result<Vec<O>, Error>
437 where
438 for<'de> O: Deserialize<'de>,
439 {
440 let file_name = path
441 .file_name()
442 .and_then(|f| f.to_str())
443 .unwrap_or("invalid_file_name")
444 .to_string();
445 if path.exists() {
446 File::open(path)
447 .map_err(|e| Error::NamedFileIO {
448 file_name: file_name.to_owned(),
449 source: Box::new(e),
450 })
451 .and_then(|r| self.read_objs(r, &file_name))
452 } else {
453 Err(Error::MissingFile(file_name))
454 }
455 }
456
457 fn read_objs_from_optional_path<O>(
458 &self,
459 dir_path: &std::path::Path,
460 file_name: &str,
461 ) -> Option<Result<Vec<O>, Error>>
462 where
463 for<'de> O: Deserialize<'de>,
464 {
465 File::open(dir_path.join(file_name))
466 .ok()
467 .map(|r| self.read_objs(r, file_name))
468 }
469
470 fn read_file<O, T>(
471 &self,
472 file_mapping: &HashMap<&&str, usize>,
473 archive: &mut zip::ZipArchive<T>,
474 file_name: &str,
475 ) -> Result<Vec<O>, Error>
476 where
477 for<'de> O: Deserialize<'de>,
478 T: std::io::Read + std::io::Seek,
479 {
480 self.read_optional_file(file_mapping, archive, file_name)
481 .unwrap_or_else(|| Err(Error::MissingFile(file_name.to_owned())))
482 }
483
484 fn read_optional_file<O, T>(
485 &self,
486 file_mapping: &HashMap<&&str, usize>,
487 archive: &mut zip::ZipArchive<T>,
488 file_name: &str,
489 ) -> Option<Result<Vec<O>, Error>>
490 where
491 for<'de> O: Deserialize<'de>,
492 T: std::io::Read + std::io::Seek,
493 {
494 file_mapping.get(&file_name).map(|i| {
495 self.read_objs(
496 archive.by_index(*i).map_err(|e| Error::NamedFileIO {
497 file_name: file_name.to_owned(),
498 source: Box::new(e),
499 })?,
500 file_name,
501 )
502 })
503 }
504}