use crate::model::{ Bilingual, EndNotice, LineSequence, Passage, RouteInfo, StopInfo, VehicleKind, VehiclePosition, }; use crate::provider::RawFeed; use crate::storage::{ DisruptionsBatch, PositionsBatch, StaticBatch, Store, TrainDeparturesBatch, TrainStationsBatch, VilloDiscoveryBatch, VilloStatusBatch, WaitingBatch, }; use std::collections::HashMap; fn bl(value: &str) -> Bilingual { Bilingual { fr: value.into(), nl: value.into(), } } const VILLO_DISCOVERY_BODY: &str = r#"{"data":{"feeds":[ {"name":"station_information","url":"https://villo.test/information"}, {"name":"station_status","url":"https://villo.test/status"} ]}}"#; const VILLO_INFORMATION_BODY: &str = r#"{"data":{"stations":[ {"station_id":"1","name":"DANSAERT","lat":50.8317,"lon":4.3856}, {"station_id":"2","name":"DE BROUCKERE","lat":50.8330,"lon":4.3865}, {"station_id":"3","name":"LAEKEN","lat":50.8300,"lon":4.3840}, {"station_id":"4","name":"BOURSE","lat":50.8322,"lon":4.3835}, {"station_id":"5","name":"ARTS-LOI","lat":50.8467,"lon":4.3676} ]}}"#; const VILLO_STATUS_BODY: &str = r#"{"data":{"stations":[ {"station_id":"1","num_vehicles_available":33,"num_docks_available":2,"vehicle_types_available":[{"vehicle_type_id":"mechanical","count":31},{"vehicle_type_id":"electrical","count":2}]}, {"station_id":"2","num_vehicles_available":18,"num_docks_available":25,"vehicle_types_available":[{"vehicle_type_id":"mechanical","count":18},{"vehicle_type_id":"electrical","count":0}]}, {"station_id":"3","num_vehicles_available":8,"num_docks_available":12,"vehicle_types_available":[{"vehicle_type_id":"mechanical","count":6},{"vehicle_type_id":"electrical","count":2}]}, {"station_id":"4","num_vehicles_available":0,"num_docks_available":14,"vehicle_types_available":[{"vehicle_type_id":"mechanical","count":0},{"vehicle_type_id":"electrical","count":0}]}, {"station_id":"5","num_vehicles_available":5,"num_docks_available":5} ]}}"#; const TRAIN_STATIONS_BODY: &str = r#"{"station":[ {"id":"BE.NMBS.008812005","name":"BRUXELLES-NORD","standardname":"BRUXELLES-NORD","locationX":"4.3856","locationY":"50.8317"}, {"id":"BE.NMBS.008813003","name":"BRUXELLES-CENTRAL","standardname":"BRUXELLES-CENTRAL","locationX":"4.3880","locationY":"50.8340"}, {"id":"BE.NMBS.008814001","name":"BRUXELLES-MIDI","standardname":"BRUXELLES-MIDI","locationX":"4.3820","locationY":"50.8290"} ]}"#; const TRAIN_STATION_IDS: [&str; 3] = [ "BE.NMBS.008812005", "BE.NMBS.008813003", "BE.NMBS.008814001", ]; fn bmc_realtime_batches( captured_at: chrono::DateTime, ) -> (WaitingBatch, PositionsBatch, DisruptionsBatch) { let realtime_raw = |feed: &'static str| RawFeed { provider: "bmc", feed, scope: "".into(), captured_at, source_at: None, content_type: "application/json".into(), body: br#"{"results":[]}"#.to_vec(), }; let expected = captured_at.fixed_offset(); let mut passages: HashMap, Vec> = HashMap::new(); let mut end_notices: HashMap, Vec> = HashMap::new(); let mut positions: HashMap, Vec> = HashMap::new(); for i in 0usize..3000 { let id = format!("S{i}"); if i >= 1500 { if i < 2000 { end_notices.insert( id.into_boxed_str(), vec![EndNotice { line: format!("L{}", i % 96), message: bl("LAST BUS HAS GONE"), }], ); } continue; } let line_a = format!("L{}", i % 96); let line_b = format!("L{}", (i + 7) % 96); passages.insert( id.clone().into_boxed_str(), vec![ Passage { line: line_a.clone(), destination: bl("CENTER"), expected: expected + chrono::Duration::seconds(60 + (i % 240) as i64), }, Passage { line: line_a.clone(), destination: bl("CENTER"), expected: expected + chrono::Duration::seconds(360 + (i % 240) as i64), }, Passage { line: line_b.clone(), destination: bl("RING"), expected: expected + chrono::Duration::seconds(120 + (i % 180) as i64), }, ], ); for (line, offset) in [(line_a, 1usize), (line_b, 2usize)] { let previous = format!("S{}", i.saturating_sub(offset)); let next = format!("S{}", (i + offset).min(2999)); positions .entry(line.into_boxed_str()) .or_default() .push(VehiclePosition { terminus: next, point_id: previous, distance_m: 500, }); } if i % 3 == 0 { end_notices.insert( id.into_boxed_str(), vec![EndNotice { line: format!("L{}", (i + 13) % 96), message: bl("LAST BUS HAS GONE"), }], ); } } ( WaitingBatch { raw: realtime_raw("waiting"), passages, end_notices, }, PositionsBatch { raw: realtime_raw("positions"), positions, }, DisruptionsBatch { raw: realtime_raw("disruptions"), disruptions: Vec::new(), }, ) } fn villo_raw( captured_at: chrono::DateTime, feed: &'static str, scope: &str, body: &str, ) -> RawFeed { RawFeed { provider: "villo", feed, scope: scope.into(), captured_at, source_at: None, content_type: "application/json".into(), body: body.as_bytes().to_vec(), } } fn villo_status_batch( captured_at: chrono::DateTime, ) -> anyhow::Result { Ok(VilloStatusBatch { raw: villo_raw(captured_at, "status", "", VILLO_STATUS_BODY), status: crate::villo::parse_status(VILLO_STATUS_BODY)?, }) } fn train_liveboard_fixture( captured_at: chrono::DateTime, ) -> anyhow::Result<(String, Vec)> { let body = format!( r#"{{"departures":{{"departure":[ {{"station":"OOSTENDE","time":"{}","delay":"120","platform":"3","vehicleinfo":{{"shortname":"IC 1","type":"IC"}}}}, {{"station":"ANVERS-CENTRAL","time":"{}","delay":"0","platform":"5","vehicleinfo":{{"shortname":"S1 1","type":"S1"}}}} ]}}}}"#, captured_at.timestamp() + 60, captured_at.timestamp() + 540, ); let departures = crate::irail::parse_liveboard(&body, captured_at.timestamp())?; Ok((body, departures)) } fn train_departures_batch( captured_at: chrono::DateTime, station_id: &str, body: &str, departures: &[crate::irail::TrainDeparture], ) -> TrainDeparturesBatch { TrainDeparturesBatch { raw: RawFeed { provider: "irail", feed: "liveboard", scope: station_id.into(), captured_at, source_at: None, content_type: "application/json".into(), body: body.as_bytes().to_vec(), }, station_id: station_id.into(), departures: departures.to_vec(), } } pub async fn refresh(store: &Store) -> anyhow::Result<()> { let captured_at = chrono::Utc::now(); let (waiting, positions, disruptions) = bmc_realtime_batches(captured_at); store.apply_waiting(waiting).await?; store.apply_positions(positions).await?; store.apply_disruptions(disruptions).await?; store .apply_villo_status(villo_status_batch(captured_at)?) .await?; let (body, departures) = train_liveboard_fixture(captured_at)?; for station_id in TRAIN_STATION_IDS { store .apply_train_departures(train_departures_batch( captured_at, station_id, &body, &departures, )) .await?; } Ok(()) } pub async fn seed(store: &Store) -> anyhow::Result<()> { let has_static = store.has_feed_head("bmc", "static").await?; let has_waiting = store.has_feed_head("bmc", "waiting").await?; let has_positions = store.has_feed_head("bmc", "positions").await?; let has_disruptions = store.has_feed_head("bmc", "disruptions").await?; let has_train_stations = store.has_feed_head("irail", "stations").await?; let mut has_train_liveboard_head = store.has_feed_head("irail", "liveboard").await?; let train_liveboard_scopes = [ ( TRAIN_STATION_IDS[0], store .has_feed_snapshot("irail", "liveboard", TRAIN_STATION_IDS[0]) .await?, ), ( TRAIN_STATION_IDS[1], store .has_feed_snapshot("irail", "liveboard", TRAIN_STATION_IDS[1]) .await?, ), ( TRAIN_STATION_IDS[2], store .has_feed_snapshot("irail", "liveboard", TRAIN_STATION_IDS[2]) .await?, ), ]; let has_all_train_liveboard_scopes = train_liveboard_scopes .iter() .all(|(_, has_snapshot)| *has_snapshot); if !has_train_liveboard_head && has_all_train_liveboard_scopes { has_train_liveboard_head = store .restore_feed_head_from_latest_snapshot("irail", "liveboard") .await? .is_some(); anyhow::ensure!(has_train_liveboard_head, "missing iRail liveboard snapshots"); } let has_train_liveboard = has_train_liveboard_head && has_all_train_liveboard_scopes; let has_villo_discovery = store.has_feed_head("villo", "discovery").await?; let has_villo_status = store.has_feed_head("villo", "status").await?; if has_static && has_waiting && has_positions && has_disruptions && has_train_stations && has_train_liveboard && has_villo_discovery && has_villo_status { return Ok(()); } let captured_at = chrono::Utc::now(); let raw_specs: [(&str, &str, &[u8]); 3] = [ ("stop-details", "application/json", br#"{"results":[]}"#), ("gtfs", "application/zip", b"mock-gtfs"), ("stops-by-line", "application/json", br#"{"results":[]}"#), ]; let raw = raw_specs .into_iter() .map(|(scope, content_type, body)| RawFeed { provider: "bmc", feed: "static", scope: scope.into(), captured_at, source_at: None, content_type: content_type.into(), body: body.to_vec(), }) .collect(); let mut stops = Vec::with_capacity(3000); let mut routes: HashMap, RouteInfo> = HashMap::new(); let mut sequences: HashMap, Vec> = HashMap::new(); let mut shapes = HashMap::new(); for line in 0..96 { let id = format!("L{line}"); routes.insert( id.clone().into_boxed_str(), RouteInfo { line: id.clone(), route_id: format!("R{line}"), kind: if line % 5 == 0 { VehicleKind::Tram } else { VehicleKind::Bus }, color: if line % 5 == 0 { "EA4F80".into() } else { "FFB178".into() }, text_color: "000000".into(), }, ); } routes.insert( "62".into(), RouteInfo { line: "62".into(), route_id: "R62-load".into(), kind: VehicleKind::Bus, color: "FFB178".into(), text_color: "000000".into(), }, ); for i in 0usize..3000 { let id = format!("S{i}"); stops.push(StopInfo { id: id.clone(), name: if i < 1500 { bl(&format!("MOCK ACTIVE {}", i % 300)) } else if i < 2000 { bl(&format!("MOCK NOTICE {}", i % 500)) } else { bl(&format!("MOCK QUIET {}", i % 1000)) }, lat: 50.80 + (i % 100) as f64 * 0.0002, lon: 4.30 + (i / 100) as f64 * 0.0002, }); if i >= 1500 { continue; } let line_a = format!("L{}", i % 96); let line_b = format!("L{}", (i + 7) % 96); for (line, destination, offset) in [ (line_a, "CENTER", 1usize), (line_b, "RING", 2usize), ] { let previous = format!("S{}", i.saturating_sub(offset)); let next = format!("S{}", (i + offset).min(2999)); sequences .entry(line.clone().into_boxed_str()) .or_default() .push(LineSequence { line: line.clone(), destination: bl(destination), stops: vec![previous.clone(), id.clone(), next.clone()], }); shapes.insert( (line, next), vec![ ( 50.80 + (i % 100) as f32 * 0.0002, 4.30 + (i / 100) as f32 * 0.0002, ), ( 50.805 + (i % 100) as f32 * 0.0002, 4.305 + (i / 100) as f32 * 0.0002, ), ( 50.810 + (i % 100) as f32 * 0.0002, 4.310 + (i / 100) as f32 * 0.0002, ), ], ); } } stops.push(StopInfo { id: "6419".into(), name: bl("MOCK LOAD 6419"), lat: 50.8076, lon: 4.3012, }); sequences.insert( "62".into(), vec![LineSequence { line: "62".into(), destination: bl("CENTER"), stops: vec!["S637".into(), "6419".into(), "S639".into()], }], ); if !has_static { store .apply_static(StaticBatch { raw, stops, routes, sequences, shapes, }) .await?; } if !has_waiting || !has_positions || !has_disruptions { let (waiting, positions, disruptions) = bmc_realtime_batches(captured_at); if !has_waiting { store.apply_waiting(waiting).await?; } if !has_positions { store.apply_positions(positions).await?; } if !has_disruptions { store.apply_disruptions(disruptions).await?; } } if !has_villo_discovery { store .apply_villo_discovery(VilloDiscoveryBatch { raw: vec![ villo_raw(captured_at, "discovery", "gbfs", VILLO_DISCOVERY_BODY), villo_raw( captured_at, "discovery", "station_information", VILLO_INFORMATION_BODY, ), ], status_url: "https://villo.test/status".into(), stations: crate::villo::parse_information(VILLO_INFORMATION_BODY)?, }) .await?; } if !has_villo_status { store .apply_villo_status(villo_status_batch(captured_at)?) .await?; } if !has_train_stations { store .apply_train_stations(TrainStationsBatch { raw: RawFeed { provider: "irail", feed: "stations", scope: "".into(), captured_at, source_at: None, content_type: "application/json".into(), body: TRAIN_STATIONS_BODY.as_bytes().to_vec(), }, stations: crate::irail::parse_stations(TRAIN_STATIONS_BODY)?, }) .await?; } if !has_train_liveboard { let (body, departures) = train_liveboard_fixture(captured_at)?; for (station_id, has_snapshot) in train_liveboard_scopes { if has_snapshot { continue; } store .apply_train_departures(train_departures_batch( captured_at, station_id, &body, &departures, )) .await?; } } Ok(()) }