GITGUD

neiam neiam /apollos_crib

It's Apollo's Crib

public · Issues · Pulls 1 · Labels · Forks · Compare · Actions failure · Badges · Packages

⭐

Log in to mark this repository.

Credo: GBFS and GTFS workers in smaller pieces

68a501b · Gabe Morell · 2026-11-04 20:55

10 files +685 -761
Message 
{commit_body(@commit)}

Files changed

modified apps/room_gbfs/lib/room_gbfs/gbfs_worker.ex
+360 −373
@@ -126,214 +126,7 @@ defmodule RoomGbfs.Worker do
126 126 case result.body
127 127 |> Jason.decode(keys: :atoms) do
128 128 {:ok, json} ->
129 − case type do
130 − :sys_info ->
131 − Repo.insert(
132 − RoomSanctum.Storage.change_sys_info(
133 − %RoomSanctum.Storage.GBFS.V1.SysInfo{},
134 − json.data
135 − |> RoomGbfs.V3.system_information(lang)
136 − |> Map.put(:source_id, id |> intify)
137 − ),
138 − on_conflict: {:replace_all_except, [:id]},
139 − conflict_target: [:source_id, :system_id]
140 − )
141 −
142 − :stat_info ->
143 − data =
144 − json.data.stations
145 − |> Enum.map(&RoomGbfs.V3.station_info(&1, lang))
146 − |> Enum.map(fn x -> inj_iddt(x, id, dt) end)
147 − |> Enum.map(fn x ->
148 − point = %Geo.Point{coordinates: {x.lon, x.lat}, srid: 4326}
149 − x |> Map.put(:place, point)
150 − end)
151 − |> Enum.map(fn x ->
152 − RoomSanctum.Storage.change_station_info(
153 − %RoomSanctum.Storage.GBFS.V1.StationInfo{},
154 − x
155 − ).changes
156 − |> inj_iddt(id, dt)
157 − end)
158 −
159 − Repo.insert_all(
160 − RoomSanctum.Storage.GBFS.V1.StationInfo,
161 − data,
162 − on_conflict: {:replace_all_except, [:id]},
163 − conflict_target: [:source_id, :station_id]
164 − )
165 −
166 − :stat_status ->
167 − # v3 dropped the electric count; the types stored from
168 − # vehicle_types.json are what puts it back.
169 − electric =
170 − id
171 − |> intify()
172 − |> RoomSanctum.Storage.gbfs_vehicle_types()
173 − |> RoomGbfs.V3.electric_type_ids()
174 −
175 − data =
176 − json.data.stations
177 − |> Enum.map(&RoomGbfs.V3.station_status(&1, electric))
178 − |> Enum.map(fn x -> inj_iddt(x, id, dt) end)
179 − |> Enum.map(fn x ->
180 − RoomSanctum.Storage.change_station_status(
181 − %RoomSanctum.Storage.GBFS.V1.StationStatus{},
182 − x
183 − ).changes
184 − |> inj_iddt(id, dt)
185 − end)
186 −
187 − Repo.insert_all(
188 − RoomSanctum.Storage.GBFS.V1.StationStatus,
189 − data,
190 − on_conflict: {:replace_all_except, [:id]},
191 − conflict_target: [:source_id, :station_id]
192 − )
193 −
194 − :free_bike ->
195 − # v3 calls them vehicles and puts them under `vehicles`.
196 − data =
197 − (Map.get(json.data, :bikes) || Map.get(json.data, :vehicles) || [])
198 − |> Enum.map(&RoomGbfs.V3.vehicle_status/1)
199 − |> Enum.map(fn x -> inj_iddt(x, id, dt) end)
200 − |> Enum.map(fn x ->
201 − point = %Geo.Point{coordinates: {x.lon, x.lat}, srid: 4326}
202 − x |> Map.put(:point, point)
203 − end)
204 − |> Enum.map(fn x ->
205 − RoomSanctum.Storage.change_free_bike_status(
206 − %RoomSanctum.Storage.GBFS.V1.FreeBikeStatus{},
207 − x
208 − ).changes
209 − |> inj_iddt(id, dt)
210 − end)
211 −
212 − RoomSanctum.Storage.truncate_free_bike_status(id)
213 −
214 − Repo.insert_all(
215 − RoomSanctum.Storage.GBFS.V1.FreeBikeStatus,
216 − data,
217 − on_conflict: {:replace_all_except, [:id]},
218 − conflict_target: [:source_id, :bike_id]
219 − )
220 −
221 − :vehicle_types ->
222 − data =
223 − json.data.vehicle_types
224 − |> Enum.map(&RoomGbfs.V3.vehicle_types(&1, lang))
225 − |> Enum.map(fn x -> inj_iddt(x, id, dt) end)
226 − |> Enum.map(fn x ->
227 − RoomSanctum.Storage.change_vehicle_types(
228 − %RoomSanctum.Storage.GBFS.V1.VehicleTypes{},
229 − x
230 − ).changes
231 − |> inj_iddt(id, dt)
232 − end)
233 −
234 − RoomSanctum.Storage.truncate_vehicle_types(id)
235 −
236 − Repo.insert_all(
237 − RoomSanctum.Storage.GBFS.V1.VehicleTypes,
238 − data,
239 − on_conflict: {:replace_all_except, [:id]},
240 − conflict_target: [:source_id, :vehicle_type_id]
241 − )
242 −
243 − :pricing_plans ->
244 − data =
245 − json.data.plans
246 − |> Enum.map(fn x -> inj_iddt(x, id, dt) end)
247 − |> Enum.map(fn x ->
248 − RoomSanctum.Storage.change_system_pricing_plans(
249 − %RoomSanctum.Storage.GBFS.V1.SystemPricingPlans{},
250 − x
251 − ).changes
252 − |> inj_iddt(id, dt)
253 − end)
254 −
255 − RoomSanctum.Storage.truncate_system_pricing_plans(id)
256 −
257 − Repo.insert_all(
258 − RoomSanctum.Storage.GBFS.V1.SystemPricingPlans,
259 − data,
260 − on_conflict: {:replace_all_except, [:id]},
261 − conflict_target: [:source_id, :plan_id]
262 − )
263 −
264 − :geofencing_zones ->
265 − data =
266 − json.data.geofencing_zones.features
267 − |> Enum.map(fn x -> inj_iddt(x, id, dt) end)
268 − |> Enum.map(fn x ->
269 − point = %Geo.MultiPolygon{coordinates: x.geometry.coordinates, srid: 4326}
270 − x |> Map.put(:place, point)
271 − end)
272 − |> Enum.map(fn x ->
273 − RoomSanctum.Storage.change_geo_fencing_zones(
274 − %RoomSanctum.Storage.GBFS.V1.GeoFencingZones{},
275 − x
276 − ).changes
277 − |> inj_iddt(id, dt)
278 − end)
279 −
280 − RoomSanctum.Storage.truncate_geo_fencing_zones(id)
281 −
282 − # Repo.insert_all(
283 − # RoomSanctum.Storage.GBFS.V1.GeoFencingZones,
284 − # data
285 − # )
286 −
287 − :sys_hours ->
288 − :ok
289 −
290 − :sys_cal ->
291 − :ok
292 −
293 − :sys_regions ->
294 − :ok
295 −
296 − :sys_alerts ->
297 − data =
298 − json.data.alerts
299 − |> Enum.map(fn x -> inj_iddt(x, id, dt) end)
300 − |> Enum.map(fn x ->
301 − RoomSanctum.Storage.change_gbfs_alert(%RoomSanctum.Storage.GBFS.V1.Alert{}, x).changes
302 − |> inj_iddt(id, dt)
303 − end)
304 −
305 − Repo.insert_all(
306 − RoomSanctum.Storage.GBFS.V1.Alert,
307 − data,
308 − on_conflict: {:replace_all_except, [:id]},
309 − conflict_target: [:source_id, :alert_id]
310 − )
311 −
312 − :ebikes_at_stations ->
313 − data =
314 − json.data.stations
315 − |> Enum.map(fn x -> inj_iddt(x, id, dt) end)
316 − |> Enum.map(fn x ->
317 − cs =
318 − RoomSanctum.Storage.change_ebikes_at_stations(
319 − %RoomSanctum.Storage.GBFS.V1.EbikesAtStations{},
320 − x
321 − ).changes
322 − |> inj_iddt(id, dt)
323 −
324 − eb = cs.ebikes |> Enum.map(fn eb -> eb |> Ecto.Changeset.apply_changes() end)
325 − cs |> Map.put(:ebikes, eb)
326 − end)
327 −
328 − RoomSanctum.Storage.truncate_ebikes_at_stations(id)
329 −
330 − Repo.insert_all(
331 − RoomSanctum.Storage.GBFS.V1.EbikesAtStations,
332 − data,
333 − on_conflict: {:replace_all_except, [:id]},
334 − conflict_target: [:source_id, :station_id]
335 − )
336 − end
129 + write_feed(type, json, id, dt, lang)
337 130
338 131 {:error, info} ->
339 132 Logger.info(info.reason)
@@ -344,6 +137,225 @@ defmodule RoomGbfs.Worker do
344 137 end
345 138 end
346 139
140 + defp write_feed(:sys_info, json, id, _dt, lang) do
141 + Repo.insert(
142 + RoomSanctum.Storage.change_sys_info(
143 + %RoomSanctum.Storage.GBFS.V1.SysInfo{},
144 + json.data
145 + |> RoomGbfs.V3.system_information(lang)
146 + |> Map.put(:source_id, id |> intify)
147 + ),
148 + on_conflict: {:replace_all_except, [:id]},
149 + conflict_target: [:source_id, :system_id]
150 + )
151 + end
152 +
153 + defp write_feed(:stat_info, json, id, dt, lang) do
154 + data =
155 + json.data.stations
156 + |> Enum.map(&RoomGbfs.V3.station_info(&1, lang))
157 + |> Enum.map(fn x -> inj_iddt(x, id, dt) end)
158 + |> Enum.map(fn x ->
159 + point = %Geo.Point{coordinates: {x.lon, x.lat}, srid: 4326}
160 + x |> Map.put(:place, point)
161 + end)
162 + |> Enum.map(fn x ->
163 + RoomSanctum.Storage.change_station_info(
164 + %RoomSanctum.Storage.GBFS.V1.StationInfo{},
165 + x
166 + ).changes
167 + |> inj_iddt(id, dt)
168 + end)
169 +
170 + Repo.insert_all(
171 + RoomSanctum.Storage.GBFS.V1.StationInfo,
172 + data,
173 + on_conflict: {:replace_all_except, [:id]},
174 + conflict_target: [:source_id, :station_id]
175 + )
176 + end
177 +
178 + defp write_feed(:stat_status, json, id, dt, _lang) do
179 + # v3 dropped the electric count; the types stored from
180 + # vehicle_types.json are what puts it back.
181 + electric =
182 + id
183 + |> intify()
184 + |> RoomSanctum.Storage.gbfs_vehicle_types()
185 + |> RoomGbfs.V3.electric_type_ids()
186 +
187 + data =
188 + json.data.stations
189 + |> Enum.map(&RoomGbfs.V3.station_status(&1, electric))
190 + |> Enum.map(fn x -> inj_iddt(x, id, dt) end)
191 + |> Enum.map(fn x ->
192 + RoomSanctum.Storage.change_station_status(
193 + %RoomSanctum.Storage.GBFS.V1.StationStatus{},
194 + x
195 + ).changes
196 + |> inj_iddt(id, dt)
197 + end)
198 +
199 + Repo.insert_all(
200 + RoomSanctum.Storage.GBFS.V1.StationStatus,
201 + data,
202 + on_conflict: {:replace_all_except, [:id]},
203 + conflict_target: [:source_id, :station_id]
204 + )
205 + end
206 +
207 + defp write_feed(:free_bike, json, id, dt, _lang) do
208 + # v3 calls them vehicles and puts them under `vehicles`.
209 + data =
210 + (Map.get(json.data, :bikes) || Map.get(json.data, :vehicles) || [])
211 + |> Enum.map(&RoomGbfs.V3.vehicle_status/1)
212 + |> Enum.map(fn x -> inj_iddt(x, id, dt) end)
213 + |> Enum.map(fn x ->
214 + point = %Geo.Point{coordinates: {x.lon, x.lat}, srid: 4326}
215 + x |> Map.put(:point, point)
216 + end)
217 + |> Enum.map(fn x ->
218 + RoomSanctum.Storage.change_free_bike_status(
219 + %RoomSanctum.Storage.GBFS.V1.FreeBikeStatus{},
220 + x
221 + ).changes
222 + |> inj_iddt(id, dt)
223 + end)
224 +
225 + RoomSanctum.Storage.truncate_free_bike_status(id)
226 +
227 + Repo.insert_all(
228 + RoomSanctum.Storage.GBFS.V1.FreeBikeStatus,
229 + data,
230 + on_conflict: {:replace_all_except, [:id]},
231 + conflict_target: [:source_id, :bike_id]
232 + )
233 + end
234 +
235 + defp write_feed(:vehicle_types, json, id, dt, lang) do
236 + data =
237 + json.data.vehicle_types
238 + |> Enum.map(&RoomGbfs.V3.vehicle_types(&1, lang))
239 + |> Enum.map(fn x -> inj_iddt(x, id, dt) end)
240 + |> Enum.map(fn x ->
241 + RoomSanctum.Storage.change_vehicle_types(
242 + %RoomSanctum.Storage.GBFS.V1.VehicleTypes{},
243 + x
244 + ).changes
245 + |> inj_iddt(id, dt)
246 + end)
247 +
248 + RoomSanctum.Storage.truncate_vehicle_types(id)
249 +
250 + Repo.insert_all(
251 + RoomSanctum.Storage.GBFS.V1.VehicleTypes,
252 + data,
253 + on_conflict: {:replace_all_except, [:id]},
254 + conflict_target: [:source_id, :vehicle_type_id]
255 + )
256 + end
257 +
258 + defp write_feed(:pricing_plans, json, id, dt, _lang) do
259 + data =
260 + json.data.plans
261 + |> Enum.map(fn x -> inj_iddt(x, id, dt) end)
262 + |> Enum.map(fn x ->
263 + RoomSanctum.Storage.change_system_pricing_plans(
264 + %RoomSanctum.Storage.GBFS.V1.SystemPricingPlans{},
265 + x
266 + ).changes
267 + |> inj_iddt(id, dt)
268 + end)
269 +
270 + RoomSanctum.Storage.truncate_system_pricing_plans(id)
271 +
272 + Repo.insert_all(
273 + RoomSanctum.Storage.GBFS.V1.SystemPricingPlans,
274 + data,
275 + on_conflict: {:replace_all_except, [:id]},
276 + conflict_target: [:source_id, :plan_id]
277 + )
278 + end
279 +
280 + defp write_feed(:geofencing_zones, json, id, dt, _lang) do
281 + data =
282 + json.data.geofencing_zones.features
283 + |> Enum.map(fn x -> inj_iddt(x, id, dt) end)
284 + |> Enum.map(fn x ->
285 + point = %Geo.MultiPolygon{coordinates: x.geometry.coordinates, srid: 4326}
286 + x |> Map.put(:place, point)
287 + end)
288 + |> Enum.map(fn x ->
289 + RoomSanctum.Storage.change_geo_fencing_zones(
290 + %RoomSanctum.Storage.GBFS.V1.GeoFencingZones{},
291 + x
292 + ).changes
293 + |> inj_iddt(id, dt)
294 + end)
295 +
296 + RoomSanctum.Storage.truncate_geo_fencing_zones(id)
297 +
298 + # Repo.insert_all(
299 + # RoomSanctum.Storage.GBFS.V1.GeoFencingZones,
300 + # data
301 + # )
302 + end
303 +
304 + defp write_feed(:sys_hours, _json, _id, _dt, _lang) do
305 + :ok
306 + end
307 +
308 + defp write_feed(:sys_cal, _json, _id, _dt, _lang) do
309 + :ok
310 + end
311 +
312 + defp write_feed(:sys_regions, _json, _id, _dt, _lang) do
313 + :ok
314 + end
315 +
316 + defp write_feed(:sys_alerts, json, id, dt, _lang) do
317 + data =
318 + json.data.alerts
319 + |> Enum.map(fn x -> inj_iddt(x, id, dt) end)
320 + |> Enum.map(fn x ->
321 + RoomSanctum.Storage.change_gbfs_alert(%RoomSanctum.Storage.GBFS.V1.Alert{}, x).changes
322 + |> inj_iddt(id, dt)
323 + end)
324 +
325 + Repo.insert_all(
326 + RoomSanctum.Storage.GBFS.V1.Alert,
327 + data,
328 + on_conflict: {:replace_all_except, [:id]},
329 + conflict_target: [:source_id, :alert_id]
330 + )
331 + end
332 +
333 + defp write_feed(:ebikes_at_stations, json, id, dt, _lang) do
334 + data =
335 + json.data.stations
336 + |> Enum.map(fn x -> inj_iddt(x, id, dt) end)
337 + |> Enum.map(fn x ->
338 + cs =
339 + RoomSanctum.Storage.change_ebikes_at_stations(
340 + %RoomSanctum.Storage.GBFS.V1.EbikesAtStations{},
341 + x
342 + ).changes
343 + |> inj_iddt(id, dt)
344 +
345 + eb = cs.ebikes |> Enum.map(fn eb -> eb |> Ecto.Changeset.apply_changes() end)
346 + cs |> Map.put(:ebikes, eb)
347 + end)
348 +
349 + RoomSanctum.Storage.truncate_ebikes_at_stations(id)
350 +
351 + Repo.insert_all(
352 + RoomSanctum.Storage.GBFS.V1.EbikesAtStations,
353 + data,
354 + on_conflict: {:replace_all_except, [:id]},
355 + conflict_target: [:source_id, :station_id]
356 + )
357 + end
358 +
347 359 def handle_call(_msg, _from, state) do
348 360 {:reply, :ok, state}
349 361 end
@@ -356,6 +368,23 @@ defmodule RoomGbfs.Worker do
356 368
357 369 @tfh 14
358 370
371 + # Each discovery feed this worker writes, in v2's names: what write_data/4
372 + # calls it, the event broadcast once it is written, and its step of @tfh.
373 + @static_feeds %{
374 + "ebikes_at_stations" => {:ebikes_at_stations, :system_information, 3},
375 + "system_information" => {:sys_info, :system_information, 4},
376 + "station_information" => {:stat_info, :system_information, 5},
377 + "station_status" => {:stat_status, :system_information, 6},
378 + "free_bike_status" => {:free_bike, :system_information, 7},
379 + "system_hours" => {:sys_hours, :system_information, 8},
380 + "system_calendar" => {:sys_cal, :system_information, 9},
381 + "system_regions" => {:sys_regions, :system_information, 10},
382 + "system_alerts" => {:sys_alerts, :system_alerts, 11},
383 + "vehicle_types" => {:vehicle_types, :vehicle_types, 12},
384 + "system_pricing_plans" => {:pricing_plans, :pricing_plans, 13},
385 + "geofencing_zones" => {:geofencing_zones, :geofencing_zones, 14}
386 + }
387 +
359 388 @impl true
360 389 def handle_cast(:update_static, state) do
361 390 cfg = Configuration.get_source!(state.id)
@@ -364,111 +393,8 @@ defmodule RoomGbfs.Worker do
364 393 true ->
365 394 Logger.info("GBFS::#{state.id} updating static info ")
366 395 bcast(state.id, :downloading, 1, @tfh)
367 −
368 − case HTTPoison.get(cfg.config.url, [], follow_redirect: true) do
369 − {:ok, result} ->
370 − # decode! here took the whole worker down when the body was not JSON,
371 − # which is exactly what an unfollowed redirect hands you.
372 − json_body =
373 − case Jason.decode(result.body) do
374 − {:ok, decoded} ->
375 − decoded
376 −
377 − {:error, _} ->
378 − Logger.error(
379 − "GBFS::#{state.id} discovery returned non-JSON (HTTP #{result.status_code}) from #{cfg.config.url}"
380 − )
381 −
382 − %{}
383 − end
384 −
385 − case RoomGbfs.V3.feeds(json_body, cfg.config.lang) do
386 − {:ok, version, feeds} ->
387 − if version == :v3 do
388 − Logger.info("GBFS::#{state.id} discovery is v3")
389 − end
390 −
391 − bcast(state.id, :parsing, 2, @tfh)
392 −
393 − feeds
394 − # Vehicle types first: station_status needs them stored to
395 − # work out how many of its bikes are electric, and operators
396 − # list the feeds in whatever order suits them.
397 − |> Enum.sort_by(&RoomGbfs.V3.feed_order/1)
398 − |> Enum.each(fn %{"name" => name, "url" => url} ->
399 − # v3 renamed the loose-vehicle feed; everything below is
400 − # written in v2's names, which is the one dialect this worker
401 − # speaks.
402 − case RoomGbfs.V3.feed_name(name) do
403 − "ebikes_at_stations" ->
404 − write_data(url, :ebikes_at_stations, state.id, cfg.config.lang)
405 − bcast(state.id, :system_information, 3, @tfh)
406 −
407 − "system_information" ->
408 − write_data(url, :sys_info, state.id, cfg.config.lang)
409 − bcast(state.id, :system_information, 4, @tfh)
410 −
411 − "station_information" ->
412 − write_data(url, :stat_info, state.id, cfg.config.lang)
413 − bcast(state.id, :system_information, 5, @tfh)
414 −
415 − "station_status" ->
416 − write_data(url, :stat_status, state.id, cfg.config.lang)
417 − bcast(state.id, :system_information, 6, @tfh)
418 −
419 − "free_bike_status" ->
420 − write_data(url, :free_bike, state.id, cfg.config.lang)
421 − bcast(state.id, :system_information, 7, @tfh)
422 −
423 − "system_hours" ->
424 − write_data(url, :sys_hours, state.id, cfg.config.lang)
425 − bcast(state.id, :system_information, 8, @tfh)
426 −
427 − "system_calendar" ->
428 − write_data(url, :sys_cal, state.id, cfg.config.lang)
429 − bcast(state.id, :system_information, 9, @tfh)
430 −
431 − "system_regions" ->
432 − write_data(url, :sys_regions, state.id, cfg.config.lang)
433 − bcast(state.id, :system_information, 10, @tfh)
434 −
435 − "system_alerts" ->
436 − write_data(url, :sys_alerts, state.id, cfg.config.lang)
437 − bcast(state.id, :system_alerts, 11, @tfh)
438 −
439 − "vehicle_types" ->
440 − write_data(url, :vehicle_types, state.id, cfg.config.lang)
441 − bcast(state.id, :vehicle_types, 12, @tfh)
442 −
443 − "system_pricing_plans" ->
444 − write_data(url, :pricing_plans, state.id, cfg.config.lang)
445 − bcast(state.id, :pricing_plans, 13, @tfh)
446 −
447 − "geofencing_zones" ->
448 − write_data(url, :geofencing_zones, state.id, cfg.config.lang)
449 − bcast(state.id, :geofencing_zones, 14, @tfh)
450 −
451 − otherwise ->
452 − Logger.info("GBFS::#{state.id} System has unhandled file #{otherwise}")
453 − end
454 − end)
455 −
456 − {:noreply, state}
457 −
458 − {:error, languages} ->
459 − bcast(
460 − state.id,
461 − :language_error,
462 − "Invalid Language Selected, available #{Enum.join(languages, ", ")}, selected: #{cfg.config.lang}"
463 − )
464 −
465 − {:noreply, state}
466 − end
467 −
468 − {:error, info} ->
469 − Logger.info(info.reason)
470 − {:noreply, state}
471 − end
396 + update_static(cfg, state.id)
397 + {:noreply, state}
472 398
473 399 false ->
474 400 bcast(state.id, :disabled)
@@ -484,77 +410,138 @@ defmodule RoomGbfs.Worker do
484 410 true ->
485 411 Logger.info("GBFS::#{state.id} updating realtime info ")
486 412 bcast(state.id, :downloading, 1, 5)
487 −
488 − case HTTPoison.get(cfg.config.url, [], follow_redirect: true) do
489 − {:ok, result} ->
490 − case result.body |> Jason.decode() do
491 − {:ok, json_body} ->
492 − case RoomGbfs.V3.feeds(json_body, cfg.config.lang) do
493 − {:ok, _version, feeds} ->
494 − bcast(state.id, :parsing, 2, 5)
495 −
496 − feeds
497 − |> Enum.sort_by(&RoomGbfs.V3.feed_order/1)
498 − |> Enum.each(fn %{"name" => name, "url" => url} ->
499 − case RoomGbfs.V3.feed_name(name) do
500 − "station_information" ->
501 − write_data(url, :stat_info, state.id, cfg.config.lang)
502 − bcast(state.id, :system_information, 3, 5)
503 −
504 − # The counts. Fetched here rather than only on the
505 − # daily static pass, which is how a dock could report
506 − # yesterday's bikes for a whole day.
507 − "station_status" ->
508 − write_data(url, :stat_status, state.id, cfg.config.lang)
509 − bcast(state.id, :system_information, 4, 5)
510 −
511 − "ebikes_at_stations" ->
512 − write_data(url, :ebikes_at_stations, state.id, cfg.config.lang)
513 − bcast(state.id, :system_information, 4, 5)
514 −
515 − "free_bike_status" ->
516 − write_data(url, :free_bike, state.id, cfg.config.lang)
517 − bcast(state.id, :system_information, 5, 5)
518 −
519 − _otherwise ->
520 − :ok
521 − end
522 − end)
523 −
524 − {:noreply, state}
525 −
526 − {:error, languages} ->
527 − bcast(
528 − state.id,
529 − :language_error,
530 − "Invalid Language Selected, available #{Enum.join(languages, ", ")}, selected: #{cfg.config.lang}"
531 − )
532 −
533 − {:noreply, state}
534 − end
535 −
536 − {:error, error} ->
537 − Logger.info(error.data)
538 − {:noreply, state}
539 − end
540 −
541 − {:error, error} ->
542 − Logger.info(error.reason)
543 − {:noreply, state}
544 − end
413 + update_realtime(cfg, state.id)
414 + {:noreply, state}
545 415
546 416 false ->
547 417 bcast(state.id, :disabled)
548 418 {:noreply, state}
549 419 end
550 −
551 − {:noreply, state}
552 420 end
553 421
554 422 def handle_cast(_msg, state) do
555 423 {:noreply, state}
556 424 end
557 425
426 + defp update_static(cfg, id) do
427 + case HTTPoison.get(cfg.config.url, [], follow_redirect: true) do
428 + {:ok, result} ->
429 + result
430 + |> discovery_json(id, cfg.config.url)
431 + |> RoomGbfs.V3.feeds(cfg.config.lang)
432 + |> write_static_feeds(id, cfg.config.lang)
433 +
434 + {:error, info} ->
435 + Logger.info(info.reason)
436 + end
437 + end
438 +
439 + # decode! here took the whole worker down when the body was not JSON,
440 + # which is exactly what an unfollowed redirect hands you.
441 + defp discovery_json(result, id, url) do
442 + case Jason.decode(result.body) do
443 + {:ok, decoded} ->
444 + decoded
445 +
446 + {:error, _} ->
447 + Logger.error(
448 + "GBFS::#{id} discovery returned non-JSON (HTTP #{result.status_code}) from #{url}"
449 + )
450 +
451 + %{}
452 + end
453 + end
454 +
455 + defp write_static_feeds({:ok, version, feeds}, id, lang) do
456 + if version == :v3 do
457 + Logger.info("GBFS::#{id} discovery is v3")
458 + end
459 +
460 + bcast(id, :parsing, 2, @tfh)
461 +
462 + feeds
463 + # Vehicle types first: station_status needs them stored to
464 + # work out how many of its bikes are electric, and operators
465 + # list the feeds in whatever order suits them.
466 + |> Enum.sort_by(&RoomGbfs.V3.feed_order/1)
467 + |> Enum.each(&write_static_feed(&1, id, lang))
468 + end
469 +
470 + defp write_static_feeds({:error, languages}, id, lang), do: language_error(id, languages, lang)
471 +
472 + defp write_static_feed(%{"name" => name, "url" => url}, id, lang) do
473 + # v3 renamed the loose-vehicle feed; everything below is
474 + # written in v2's names, which is the one dialect this worker
475 + # speaks.
476 + case Map.fetch(@static_feeds, RoomGbfs.V3.feed_name(name)) do
477 + {:ok, {type, event, step}} ->
478 + write_data(url, type, id, lang)
479 + bcast(id, event, step, @tfh)
480 +
481 + :error ->
482 + Logger.info("GBFS::#{id} System has unhandled file #{RoomGbfs.V3.feed_name(name)}")
483 + end
484 + end
485 +
486 + # The feeds the five minute pass refreshes, and each one's step of five.
487 + @realtime_feeds %{
488 + "station_information" => {:stat_info, 3},
489 + # The counts. Fetched here rather than only on the
490 + # daily static pass, which is how a dock could report
491 + # yesterday's bikes for a whole day.
492 + "station_status" => {:stat_status, 4},
493 + "ebikes_at_stations" => {:ebikes_at_stations, 4},
494 + "free_bike_status" => {:free_bike, 5}
495 + }
496 +
497 + defp update_realtime(cfg, id) do
498 + case HTTPoison.get(cfg.config.url, [], follow_redirect: true) do
499 + {:ok, result} ->
500 + case result.body |> Jason.decode() do
501 + {:ok, json_body} ->
502 + json_body
503 + |> RoomGbfs.V3.feeds(cfg.config.lang)
504 + |> write_realtime_feeds(id, cfg.config.lang)
505 +
506 + {:error, error} ->
507 + Logger.info(error.data)
508 + end
509 +
510 + {:error, error} ->
511 + Logger.info(error.reason)
512 + end
513 + end
514 +
515 + defp write_realtime_feeds({:ok, _version, feeds}, id, lang) do
516 + bcast(id, :parsing, 2, 5)
517 +
518 + feeds
519 + |> Enum.sort_by(&RoomGbfs.V3.feed_order/1)
520 + |> Enum.each(&write_realtime_feed(&1, id, lang))
521 + end
522 +
523 + defp write_realtime_feeds({:error, languages}, id, lang),
524 + do: language_error(id, languages, lang)
525 +
526 + defp write_realtime_feed(%{"name" => name, "url" => url}, id, lang) do
527 + case Map.fetch(@realtime_feeds, RoomGbfs.V3.feed_name(name)) do
528 + {:ok, {type, step}} ->
529 + write_data(url, type, id, lang)
530 + bcast(id, :system_information, step, 5)
531 +
532 + :error ->
533 + :ok
534 + end
535 + end
536 +
537 + defp language_error(id, languages, lang) do
538 + bcast(
539 + id,
540 + :language_error,
541 + "Invalid Language Selected, available #{Enum.join(languages, ", ")}, selected: #{lang}"
542 + )
543 + end
544 +
558 545 defp via_tuple(name), do: {:via, Registry, {@registry, name}}
559 546
560 547 def sys_info_as_stats(id) do
modified apps/room_gtfs/lib/room_gtfs/feed_tester.ex
+23 −15
@@ -260,21 +260,29 @@ defmodule RoomGtfs.FeedTester do
260 260 end
261 261 end
262 262
263 − defp do_check_agency(result, feed, source) do
264 − prefixes =
265 − feed.entity
266 − |> Enum.map(&trip_id/1)
267 − |> Enum.reject(&is_nil/1)
268 − |> Enum.map(fn id ->
269 − case String.split(id, ":", parts: 2) do
270 − [prefix, _rest] -> prefix
271 − _ -> nil
272 − end
273 − end)
274 − |> Enum.reject(&is_nil/1)
275 − |> Enum.uniq()
276 − |> Enum.sort()
263 + # The operator prefixes on the feed's trip ids -- the part before the first
264 + # colon -- each once, sorted.
265 + defp trip_prefixes(feed) do
266 + feed.entity
267 + |> Enum.map(&trip_id/1)
268 + |> Enum.reject(&is_nil/1)
269 + |> Enum.map(fn id ->
270 + case String.split(id, ":", parts: 2) do
271 + [prefix, _rest] -> prefix
272 + _ -> nil
273 + end
274 + end)
275 + |> Enum.reject(&is_nil/1)
276 + |> Enum.uniq()
277 + |> Enum.sort()
278 + end
277 279
280 + defp shown_prefixes(prefixes) do
281 + "#{Enum.join(Enum.take(prefixes, 6), ", ")}#{if length(prefixes) > 6, do: ", …", else: ""}"
282 + end
283 +
284 + defp do_check_agency(result, feed, source) do
285 + prefixes = trip_prefixes(feed)
278 286 configured = Map.get(source.config, :rt_agency)
279 287 result = Map.put(result, :prefixes, prefixes)
280 288
@@ -283,7 +291,7 @@ defmodule RoomGtfs.FeedTester do
283 291 add(
284 292 result,
285 293 :error,
286 − "trip ids are prefixed (#{Enum.join(Enum.take(prefixes, 6), ", ")}#{if length(prefixes) > 6, do: ", …", else: ""}) " <>
294 + "trip ids are prefixed (#{shown_prefixes(prefixes)}) " <>
287 295 "but no operator code is set, so nothing will match the schedule"
288 296 )
289 297
modified apps/room_gtfs/lib/room_gtfs/gtfs_worker.ex
+302 −373
@@ -308,41 +308,40 @@ defmodule RoomGtfs.Worker do
308 308 vehicles = occupancy_vehicles(id, config, trips)
309 309
310 310 res
311 − |> Enum.map(fn x ->
312 − case Enum.find(rtvals, fn v ->
313 − trip_id_match?(config, x.trip_id, v.trip_id)
314 − end) do
315 − nil ->
316 − x
317 −
318 − val ->
319 − stu = val.stop
320 − time_event = stu && (stu.arrival || stu.departure)
321 −
322 − x =
323 − x
324 − |> merge_occupancy(occupancy_for(config, x.trip_id, stu, vehicles))
325 − |> merge_fleet_mark(config, vehicles)
326 − |> merge_schedule_relationship(val.schedule_relationship, stu)
327 − |> merge_stop_properties(stu)
328 −
329 − case time_event do
330 − nil ->
331 − x
332 −
333 − event ->
334 − x
335 − |> Map.put(:arrival_time_live_ts, event.time)
336 − |> Map.put(:arrival_time_live_delay, event.delay)
337 − |> Map.put(:arrival_time_live_uncertianty, event.uncertainty)
338 − end
339 − end
340 − end)
311 + |> Enum.map(&merge_realtime(&1, rtvals, config, vehicles))
341 312 |> resolve_assigned_platforms(id)
342 313 end
343 314 end
344 315 end
345 316
317 + # One scheduled arrival with whatever the trip update for its trip says.
318 + defp merge_realtime(x, rtvals, config, vehicles) do
319 + case Enum.find(rtvals, fn v -> trip_id_match?(config, x.trip_id, v.trip_id) end) do
320 + nil ->
321 + x
322 +
323 + val ->
324 + stu = val.stop
325 + time_event = stu && (stu.arrival || stu.departure)
326 +
327 + x
328 + |> merge_occupancy(occupancy_for(config, x.trip_id, stu, vehicles))
329 + |> merge_fleet_mark(config, vehicles)
330 + |> merge_schedule_relationship(val.schedule_relationship, stu)
331 + |> merge_stop_properties(stu)
332 + |> merge_live_time(time_event)
333 + end
334 + end
335 +
336 + defp merge_live_time(x, nil), do: x
337 +
338 + defp merge_live_time(x, event) do
339 + x
340 + |> Map.put(:arrival_time_live_ts, event.time)
341 + |> Map.put(:arrival_time_live_delay, event.delay)
342 + |> Map.put(:arrival_time_live_uncertianty, event.uncertainty)
343 + end
344 +
346 345 # How long to wait for vehicle positions before giving up on occupancy.
347 346 #
348 347 # Deliberately short. This is the second call query_stop makes to the same
@@ -962,47 +961,10 @@ defmodule RoomGtfs.Worker.RT do
962 961 |> Enum.filter(& &1.alert)
963 962 |> Enum.map(& &1.alert)
964 963 |> Enum.filter(fn alert ->
965 − active =
966 − case alert.active_period do
967 − [] ->
968 − true
969 −
970 − periods ->
971 − Enum.any?(periods, fn p ->
972 − (p.start == nil || p.start <= now) &&
973 − (Map.get(p, :end) == nil || Map.get(p, :end) >= now)
974 − end)
975 − end
976 −
977 − relevant =
978 − Enum.any?(alert.informed_entity, fn e ->
979 − (e.stop_id != nil && e.stop_id == stop) ||
980 − (e.route_id != nil && e.route_id in route_ids) ||
981 − (e.agency_id != nil && e.stop_id == nil && e.route_id == nil) ||
982 − (e.stop_id == nil && e.route_id == nil && e.trip == nil && e.agency_id == nil)
983 − end)
984 −
985 − active && relevant
986 − end)
987 − |> Enum.map(fn alert ->
988 − route_id = Enum.find_value(alert.informed_entity, fn e -> e.route_id end)
989 −
990 − %{
991 − effect: alert.effect |> to_string(),
992 − cause: alert.cause |> to_string(),
993 − header: get_translation(alert.header_text),
994 − description: get_translation(alert.description_text),
995 − route_id: route_id,
996 − severity: alert.severity_level |> to_string(),
997 − # Whether the alert names this stop, as opposed to reaching it by
998 − # naming the line it is on. "Elevator out here" and "delays on the
999 − # Red Line" are both true of the stop and are not the same news.
1000 − stop_specific:
1001 − Enum.any?(alert.informed_entity, fn e ->
1002 − e.stop_id != nil and e.stop_id == stop
1003 − end)
1004 − }
964 + alert_active?(alert, now) &&
965 + Enum.any?(alert.informed_entity, &reaches_stop?(&1, stop, route_ids))
1005 966 end)
967 + |> Enum.map(&present_stop_alert(&1, stop))
1006 968 end
1007 969
1008 970 {:reply, alerts, state}
@@ -1038,6 +1000,35 @@ defmodule RoomGtfs.Worker.RT do
1038 1000 end
1039 1001 end
1040 1002
1003 + # An informed entity reaches a stop by naming it, by naming a route in
1004 + # route_ids, or by naming neither -- the whole agency when it carries an
1005 + # agency id, the whole feed when it carries no trip either.
1006 + defp reaches_stop?(e, stop, route_ids) do
1007 + (e.stop_id != nil and e.stop_id == stop) or
1008 + (e.route_id != nil and e.route_id in route_ids) or
1009 + (e.stop_id == nil and e.route_id == nil and (e.agency_id != nil or e.trip == nil))
1010 + end
1011 +
1012 + defp present_stop_alert(alert, stop) do
1013 + route_id = Enum.find_value(alert.informed_entity, fn e -> e.route_id end)
1014 +
1015 + %{
1016 + effect: alert.effect |> to_string(),
1017 + cause: alert.cause |> to_string(),
1018 + header: get_translation(alert.header_text),
1019 + description: get_translation(alert.description_text),
1020 + route_id: route_id,
1021 + severity: alert.severity_level |> to_string(),
1022 + # Whether the alert names this stop, as opposed to reaching it by
1023 + # naming the line it is on. "Elevator out here" and "delays on the
1024 + # Red Line" are both true of the stop and are not the same news.
1025 + stop_specific:
1026 + Enum.any?(alert.informed_entity, fn e ->
1027 + e.stop_id != nil and e.stop_id == stop
1028 + end)
1029 + }
1030 + end
1031 +
1041 1032 # An alert names what it affects as a list of informed entities, each of
1042 1033 # which may carry a route, a stop, both, or neither -- "neither" meaning the
1043 1034 # whole agency. Those ids are the thing worth reading: they say whether an
@@ -1162,11 +1153,11 @@ defmodule RoomGtfs.Worker.RT do
1162 1153
1163 1154 defp place_vehicle(v, stops) do
1164 1155 base = %{
1165 − vehicle_id: v.vehicle && v.vehicle.id,
1156 + vehicle_id: field_of(v.vehicle, :id),
1166 1157 # What is painted on it, which for the MBTA is the car numbers.
1167 − label: v.vehicle && v.vehicle.label,
1168 − trip_id: v.trip && v.trip.trip_id,
1169 − route_id: v.trip && v.trip.route_id,
1158 + label: field_of(v.vehicle, :label),
1159 + trip_id: field_of(v.trip, :trip_id),
1160 + route_id: field_of(v.trip, :route_id),
1170 1161 stop_id: v.stop_id,
1171 1162 current_status: v.current_status,
1172 1163 occupancy_status: v.occupancy_status,
@@ -1198,6 +1189,10 @@ defmodule RoomGtfs.Worker.RT do
1198 1189 end
1199 1190 end
1200 1191
1192 + # A field of a descriptor the feed may have left out.
1193 + defp field_of(nil, _key), do: nil
1194 + defp field_of(descriptor, key), do: Map.fetch!(descriptor, key)
1195 +
1201 1196 # Positions for the stops this feed actually names, as the fallback above
1202 1197 # needs them.
1203 1198 #
@@ -1944,6 +1939,20 @@ defmodule RoomGtfs.Worker.Static do
1944 1939 )
1945 1940 end
1946 1941
1942 + # Each file type written by COPY: its schema, and what empties the source's
1943 + # rows of it first.
1944 + @copy_tables %{
1945 + agencies: {GTFS.Agency, &RoomSanctum.Storage.truncate_agency/1},
1946 + calendars: {GTFS.Calendar, &RoomSanctum.Storage.truncate_calendar/1},
1947 + calendar_dates: {GTFS.CalendarDate, &RoomSanctum.Storage.truncate_calendar_date/1},
1948 + directions: {GTFS.Direction, &RoomSanctum.Storage.truncate_direction/1},
1949 + routes: {GTFS.Route, &RoomSanctum.Storage.truncate_route/1},
1950 + stops: {GTFS.Stop, &RoomSanctum.Storage.truncate_stop/1},
1951 + stop_times: {GTFS.StopTime, &RoomSanctum.Storage.truncate_stop_time/1},
1952 + trips: {GTFS.Trip, &RoomSanctum.Storage.truncate_trip/1},
1953 + shapes: {GTFS.Shape, &RoomSanctum.Storage.truncate_shape/1}
1954 + }
1955 +
1947 1956 defp write_file(contents, type, id, pid) do
1948 1957 datetime = NaiveDateTime.local_now()
1949 1958 Logger.info("GTFS::#{id} writing #{type} (c)")
@@ -1968,58 +1977,12 @@ defmodule RoomGtfs.Worker.Static do
1968 1977 # contents |> Stream.take(500) |> Enum.to_list() |> IO.inspect
1969 1978
1970 1979 # add truncation here as necessary
1971 − case type do
1972 − :agencies -> RoomSanctum.Storage.truncate_agency(id)
1973 − :calendars -> RoomSanctum.Storage.truncate_calendar(id)
1974 − :calendar_dates -> RoomSanctum.Storage.truncate_calendar_date(id)
1975 − :directions -> RoomSanctum.Storage.truncate_direction(id)
1976 − :routes -> RoomSanctum.Storage.truncate_route(id)
1977 − :stops -> RoomSanctum.Storage.truncate_stop(id)
1978 − :stop_times -> RoomSanctum.Storage.truncate_stop_time(id)
1979 − :trips -> RoomSanctum.Storage.truncate_trip(id)
1980 − :shapes -> RoomSanctum.Storage.truncate_shape(id)
1981 − _ -> :ok
1982 − end
1980 + {schema, truncate} = Map.fetch!(@copy_tables, type)
1981 + truncate.(id)
1983 1982
1984 1983 # set our variables based on the type
1985 − {table, columns, pg_cols} =
1986 − case type do
1987 − :agencies ->
1988 − {:gtfs_agencies, [GTFS.Agency |> get_cols(cols_j_plus)],
1989 − GTFS.Agency |> get_cols_pgtypes(cols_j_plus)}
1990 −
1991 − :calendars ->
1992 − {:gtfs_calendars, [GTFS.Calendar |> get_cols(cols_j_plus)],
1993 − GTFS.Calendar |> get_cols_pgtypes(cols_j_plus)}
1994 −
1995 − :calendar_dates ->
1996 − {:gtfs_calendar_dates, [GTFS.CalendarDate |> get_cols(cols_j_plus)],
1997 − GTFS.CalendarDate |> get_cols_pgtypes(cols_j_plus)}
1998 −
1999 − :directions ->
2000 − {:gtfs_directions, [GTFS.Direction |> get_cols(cols_j_plus)],
2001 − GTFS.Direction |> get_cols_pgtypes(cols_j_plus)}
2002 −
2003 − :routes ->
2004 − {:gtfs_routes, [GTFS.Route |> get_cols(cols_j_plus)],
2005 − GTFS.Route |> get_cols_pgtypes(cols_j_plus)}
2006 −
2007 − :stops ->
2008 − {:gtfs_stops, [GTFS.Stop |> get_cols(cols_j_plus)],
2009 − GTFS.Stop |> get_cols_pgtypes(cols_j_plus)}
2010 −
2011 − :stop_times ->
2012 − {:gtfs_stop_times, [GTFS.StopTime |> get_cols(cols_j_plus)],
2013 − GTFS.StopTime |> get_cols_pgtypes(cols_j_plus)}
2014 −
2015 − :trips ->
2016 − {:gtfs_trips, [GTFS.Trip |> get_cols(cols_j_plus)],
2017 − GTFS.Trip |> get_cols_pgtypes(cols_j_plus)}
2018 −
2019 − :shapes ->
2020 − {:gtfs_shapes, [GTFS.Shape |> get_cols(cols_j_plus)],
2021 − GTFS.Shape |> get_cols_pgtypes(cols_j_plus)}
2022 − end
1984 + columns = [schema |> get_cols(cols_j_plus)]
1985 + pg_cols = schema |> get_cols_pgtypes(cols_j_plus)
2023 1986
2024 1987 # IO.inspect({table, columns, pg_cols})
2025 1988 # `Repo.config/0` carries Ecto's `timeout: 15000`, which looks like it would
@@ -2107,12 +2070,7 @@ defmodule RoomGtfs.Worker.Static do
2107 2070 datetime = NaiveDateTime.local_now()
2108 2071 Logger.info("GTFS::#{id} writing #{type}")
2109 2072
2110 − case type do
2111 − :routes -> RoomSanctum.Storage.truncate_route(id)
2112 − :stops -> RoomSanctum.Storage.truncate_stop(id)
2113 − :stop_times -> RoomSanctum.Storage.truncate_stop_time(id)
2114 − _ -> :ok
2115 − end
2073 + truncate_rows(type, id)
2116 2074
2117 2075 contents
2118 2076 |> Stream.filter(fn {status, data} -> status == :ok end)
@@ -2124,34 +2082,8 @@ defmodule RoomGtfs.Worker.Static do
2124 2082 |> Map.put("inserted_at", datetime)
2125 2083 |> Map.put("updated_at", datetime)
2126 2084
2127 − case type do
2128 − :agency ->
2129 − RoomSanctum.Storage.change_agency(%RoomSanctum.Storage.GTFS.Agency{}, relevant_data).changes
2130 −
2131 − :calendar ->
2132 − RoomSanctum.Storage.change_calendar(%RoomSanctum.Storage.GTFS.Calendar{}, relevant_data).changes
2133 −
2134 − :directions ->
2135 − RoomSanctum.Storage.change_direction(
2136 − %RoomSanctum.Storage.GTFS.Direction{},
2137 − relevant_data
2138 − ).changes
2139 −
2140 − :routes ->
2141 − RoomSanctum.Storage.change_route(%RoomSanctum.Storage.GTFS.Route{}, relevant_data).changes
2142 −
2143 − :stops ->
2144 − RoomSanctum.Storage.change_stop(%RoomSanctum.Storage.GTFS.Stop{}, relevant_data).changes
2145 −
2146 − :stop_times ->
2147 − RoomSanctum.Storage.change_stop_time(
2148 − %RoomSanctum.Storage.GTFS.StopTime{},
2149 − relevant_data
2150 − ).changes
2151 −
2152 − :trips ->
2153 − RoomSanctum.Storage.change_trip(%RoomSanctum.Storage.GTFS.Trip{}, relevant_data).changes
2154 − end
2085 + type
2086 + |> row_changes(relevant_data)
2155 2087 |> Map.put(:inserted_at, datetime)
2156 2088 |> Map.put(:updated_at, datetime)
2157 2089 end)
@@ -2161,61 +2093,64 @@ defmodule RoomGtfs.Worker.Static do
2161 2093 chunked_data
2162 2094 |> Enum.uniq()
2163 2095
2164 − case type do
2165 − :agency ->
2166 − Repo.insert_all(
2167 − RoomSanctum.Storage.GTFS.Agency,
2168 − chunked_data,
2169 − on_conflict: {:replace_all_except, [:id]},
2170 − conflict_target: [:source_id, :agency_id]
2171 − )
2096 + insert_rows(type, chunked_data)
2097 + end)
2098 + |> Stream.run()
2172 2099
2173 − :calendar ->
2174 − Repo.insert_all(
2175 − RoomSanctum.Storage.GTFS.Calendar,
2176 − chunked_data,
2177 − on_conflict: {:replace_all_except, [:id]},
2178 − conflict_target: [:source_id, :service_id]
2179 − )
2100 + DateTime.utc_now()
2101 + end
2180 2102
2181 − :directions ->
2182 − Repo.insert_all(
2183 − RoomSanctum.Storage.GTFS.Direction,
2184 − chunked_data,
2185 − on_conflict: {:replace_all_except, [:id]},
2186 − conflict_target: [:source_id, :route_id, :direction_id]
2187 − )
2103 + # The types written row by row that are emptied first rather than upserted.
2104 + defp truncate_rows(:routes, id), do: RoomSanctum.Storage.truncate_route(id)
2105 + defp truncate_rows(:stops, id), do: RoomSanctum.Storage.truncate_stop(id)
2106 + defp truncate_rows(:stop_times, id), do: RoomSanctum.Storage.truncate_stop_time(id)
2107 + defp truncate_rows(_type, _id), do: :ok
2188 2108
2189 − :routes ->
2190 − Repo.insert_all(
2191 − RoomSanctum.Storage.GTFS.Route,
2192 − chunked_data
2193 − )
2109 + defp row_changes(:agency, data),
2110 + do: RoomSanctum.Storage.change_agency(%RoomSanctum.Storage.GTFS.Agency{}, data).changes
2194 2111
2195 − :stops ->
2196 − Repo.insert_all(
2197 − RoomSanctum.Storage.GTFS.Stop,
2198 − chunked_data
2199 − )
2112 + defp row_changes(:calendar, data),
2113 + do: RoomSanctum.Storage.change_calendar(%RoomSanctum.Storage.GTFS.Calendar{}, data).changes
2200 2114
2201 − :stop_times ->
2202 − Repo.insert_all(
2203 − RoomSanctum.Storage.GTFS.StopTime,
2204 − chunked_data
2205 − )
2115 + defp row_changes(:directions, data),
2116 + do: RoomSanctum.Storage.change_direction(%RoomSanctum.Storage.GTFS.Direction{}, data).changes
2206 2117
2207 − :trips ->
2208 − Repo.insert_all(
2209 − RoomSanctum.Storage.GTFS.Trip,
2210 − chunked_data,
2211 − on_conflict: {:replace_all_except, [:id]},
2212 − conflict_target: [:source_id, :trip_id]
2213 − )
2214 − end
2215 − end)
2216 − |> Stream.run()
2118 + defp row_changes(:routes, data),
2119 + do: RoomSanctum.Storage.change_route(%RoomSanctum.Storage.GTFS.Route{}, data).changes
2217 2120
2218 − DateTime.utc_now()
2121 + defp row_changes(:stops, data),
2122 + do: RoomSanctum.Storage.change_stop(%RoomSanctum.Storage.GTFS.Stop{}, data).changes
2123 +
2124 + defp row_changes(:stop_times, data),
2125 + do: RoomSanctum.Storage.change_stop_time(%RoomSanctum.Storage.GTFS.StopTime{}, data).changes
2126 +
2127 + defp row_changes(:trips, data),
2128 + do: RoomSanctum.Storage.change_trip(%RoomSanctum.Storage.GTFS.Trip{}, data).changes
2129 +
2130 + defp insert_rows(:agency, rows),
2131 + do: upsert_rows(RoomSanctum.Storage.GTFS.Agency, rows, [:source_id, :agency_id])
2132 +
2133 + defp insert_rows(:calendar, rows),
2134 + do: upsert_rows(RoomSanctum.Storage.GTFS.Calendar, rows, [:source_id, :service_id])
2135 +
2136 + defp insert_rows(:directions, rows),
2137 + do:
2138 + upsert_rows(RoomSanctum.Storage.GTFS.Direction, rows, [:source_id, :route_id, :direction_id])
2139 +
2140 + defp insert_rows(:routes, rows), do: Repo.insert_all(RoomSanctum.Storage.GTFS.Route, rows)
2141 + defp insert_rows(:stops, rows), do: Repo.insert_all(RoomSanctum.Storage.GTFS.Stop, rows)
2142 +
2143 + defp insert_rows(:stop_times, rows),
2144 + do: Repo.insert_all(RoomSanctum.Storage.GTFS.StopTime, rows)
2145 +
2146 + defp insert_rows(:trips, rows),
2147 + do: upsert_rows(RoomSanctum.Storage.GTFS.Trip, rows, [:source_id, :trip_id])
2148 +
2149 + defp upsert_rows(schema, rows, conflict_target) do
2150 + Repo.insert_all(schema, rows,
2151 + on_conflict: {:replace_all_except, [:id]},
2152 + conflict_target: conflict_target
2153 + )
2219 2154 end
2220 2155
2221 2156 # linked_datasets.txt associates GTFS-RT feeds with the schedule that
@@ -2346,36 +2281,19 @@ defmodule RoomGtfs.Worker.Static do
2346 2281 String.starts_with?(name, "__MACOSX/") or String.starts_with?(Path.basename(name), "._")
2347 2282 end
2348 2283
2349 − defp file_to_atom(filename) do
2350 − case filename do
2351 − "agency.txt" ->
2352 − :agencies
2353 −
2354 − "calendar.txt" ->
2355 − :calendars
2356 −
2357 − "calendar_dates.txt" ->
2358 − :calendar_dates
2359 −
2360 − "directions.txt" ->
2361 − :directions
2362 −
2363 − "routes.txt" ->
2364 − :routes
2365 −
2366 − "stops.txt" ->
2367 − :stops
2368 −
2369 − "stop_times.txt" ->
2370 − :stop_times
2284 + @file_types %{
2285 + "agency.txt" => :agencies,
2286 + "calendar.txt" => :calendars,
2287 + "calendar_dates.txt" => :calendar_dates,
2288 + "directions.txt" => :directions,
2289 + "routes.txt" => :routes,
2290 + "stops.txt" => :stops,
2291 + "stop_times.txt" => :stop_times,
2292 + "trips.txt" => :trips,
2293 + "shapes.txt" => :shapes
2294 + }
2371 2295
2372 − "trips.txt" ->
2373 − :trips
2374 −
2375 − "shapes.txt" ->
2376 − :shapes
2377 − end
2378 − end
2296 + defp file_to_atom(filename), do: Map.fetch!(@file_types, filename)
2379 2297
2380 2298 # The order the progress bar counts in, which has to be the order the files are
2381 2299 # actually written in or the bar moves backwards. Entries come from the zip's
@@ -2384,19 +2302,19 @@ defmodule RoomGtfs.Worker.Static do
2384 2302 # Being numbered last meant its step arrived as "10 of 10", which the source
2385 2303 # page reads as completion: the bar hit 100% and flipped to idle while three
2386 2304 # more files, stop_times included, were still to load.
2387 − def file_to_order(filename) do
2388 − case filename do
2389 − "agency.txt" -> 3
2390 − "calendar.txt" -> 4
2391 − "calendar_dates.txt" -> 5
2392 − "directions.txt" -> 6
2393 − "routes.txt" -> 7
2394 − "shapes.txt" -> 8
2395 − "stops.txt" -> 9
2396 − "stop_times.txt" -> 10
2397 − "trips.txt" -> 11
2398 − end
2399 − end
2305 + @file_order %{
2306 + "agency.txt" => 3,
2307 + "calendar.txt" => 4,
2308 + "calendar_dates.txt" => 5,
2309 + "directions.txt" => 6,
2310 + "routes.txt" => 7,
2311 + "shapes.txt" => 8,
2312 + "stops.txt" => 9,
2313 + "stop_times.txt" => 10,
2314 + "trips.txt" => 11
2315 + }
2316 +
2317 + def file_to_order(filename), do: Map.fetch!(@file_order, filename)
2400 2318
2401 2319 defp atom_to_table(atom) do
2402 2320 "gtfs_#{atom}"
@@ -2499,60 +2417,7 @@ defmodule RoomGtfs.Worker.Static do
2499 2417 bcast(id, :extracting, 2, 11)
2500 2418 Logger.info("GTFS::#{id} downloaded #{byte_size(result.body)} bytes")
2501 2419
2502 − case result.body |> trim_zip_tail() |> Unzip.InMem.new() |> Unzip.new() do
2503 − {:ok, unzip} ->
2504 − files = Unzip.list_entries(unzip)
2505 −
2506 − case Enum.find(files, &(&1.file_name == "linked_datasets.txt")) do
2507 − nil ->
2508 − :ok
2509 −
2510 − entry ->
2511 − Unzip.file_stream!(unzip, entry.file_name)
2512 − |> Enum.to_list()
2513 − |> IO.iodata_to_binary()
2514 − |> parse_linked_datasets()
2515 − |> linked_dataset_urls()
2516 − |> then(&apply_linked_datasets(cfg, &1))
2517 − end
2518 −
2519 − try do
2520 − # Sorted, because `Unzip.list_entries/1` maps over `cd_list`, which
2521 − # is a map -- so the order it hands back is neither the zip's nor
2522 − # alphabetical, it is whatever the hash gives. Left to that, the
2523 − # progress bar walks backwards: stop_times (step 9) came out ahead
2524 − # of stops (step 8) on the MTA feeds. Sorting by the same function
2525 − # that numbers the steps makes the order the bar claims the order
2526 − # the files are actually written in, for every feed.
2527 − files
2528 − |> gtfs_entries(id)
2529 − |> Enum.sort_by(&file_to_order(Path.basename(&1.file_name)))
2530 − |> Enum.each(fn e ->
2531 − name = Path.basename(e.file_name)
2532 − type = file_to_atom(name)
2533 −
2534 − bcast(id, type, file_to_order(name), 12)
2535 −
2536 − Unzip.file_stream!(unzip, e.file_name)
2537 − |> write_file(type, id, nil)
2538 − end)
2539 − rescue
2540 − e ->
2541 − Logger.error("GTFS::#{id} error during static import: #{inspect(e)}")
2542 − after
2543 − Configuration.update_source_meta(cfg, %{last_run: DateTime.utc_now()})
2544 − bcast(id, :done)
2545 − end
2546 −
2547 − :ok
2548 −
2549 − {:error, term} ->
2550 − # `Logger.error/1` takes chardata, not a struct: passing the
2551 − # raw term raised here, inside the branch meant to report the
2552 − # error.
2553 − Logger.error("GTFS::#{id} static feed was not a readable zip: #{inspect(term)}")
2554 − {:error, {:unzip, term}}
2555 − end
2420 + import_zip(id, cfg, result.body)
2556 2421
2557 2422 # A server that answers a zip URL with an error page was, until now,
2558 2423 # reported as a feed that was not a readable zip -- true, and useless.
@@ -2572,6 +2437,66 @@ defmodule RoomGtfs.Worker.Static do
2572 2437 end
2573 2438 end
2574 2439
2440 + defp import_zip(id, cfg, body) do
2441 + case body |> trim_zip_tail() |> Unzip.InMem.new() |> Unzip.new() do
2442 + {:ok, unzip} ->
2443 + files = Unzip.list_entries(unzip)
2444 + apply_linked_datasets_entry(unzip, files, cfg)
2445 + write_entries(unzip, files, id, cfg)
2446 + :ok
2447 +
2448 + {:error, term} ->
2449 + # `Logger.error/1` takes chardata, not a struct: passing the
2450 + # raw term raised here, inside the branch meant to report the
2451 + # error.
2452 + Logger.error("GTFS::#{id} static feed was not a readable zip: #{inspect(term)}")
2453 + {:error, {:unzip, term}}
2454 + end
2455 + end
2456 +
2457 + defp apply_linked_datasets_entry(unzip, files, cfg) do
2458 + case Enum.find(files, &(&1.file_name == "linked_datasets.txt")) do
2459 + nil ->
2460 + :ok
2461 +
2462 + entry ->
2463 + Unzip.file_stream!(unzip, entry.file_name)
2464 + |> Enum.to_list()
2465 + |> IO.iodata_to_binary()
2466 + |> parse_linked_datasets()
2467 + |> linked_dataset_urls()
2468 + |> then(&apply_linked_datasets(cfg, &1))
2469 + end
2470 + end
2471 +
2472 + defp write_entries(unzip, files, id, cfg) do
2473 + # Sorted, because `Unzip.list_entries/1` maps over `cd_list`, which
2474 + # is a map -- so the order it hands back is neither the zip's nor
2475 + # alphabetical, it is whatever the hash gives. Left to that, the
2476 + # progress bar walks backwards: stop_times (step 9) came out ahead
2477 + # of stops (step 8) on the MTA feeds. Sorting by the same function
2478 + # that numbers the steps makes the order the bar claims the order
2479 + # the files are actually written in, for every feed.
2480 + files
2481 + |> gtfs_entries(id)
2482 + |> Enum.sort_by(&file_to_order(Path.basename(&1.file_name)))
2483 + |> Enum.each(fn e ->
2484 + name = Path.basename(e.file_name)
2485 + type = file_to_atom(name)
2486 +
2487 + bcast(id, type, file_to_order(name), 12)
2488 +
2489 + Unzip.file_stream!(unzip, e.file_name)
2490 + |> write_file(type, id, nil)
2491 + end)
2492 + rescue
2493 + e ->
2494 + Logger.error("GTFS::#{id} error during static import: #{inspect(e)}")
2495 + after
2496 + Configuration.update_source_meta(cfg, %{last_run: DateTime.utc_now()})
2497 + bcast(id, :done)
2498 + end
2499 +
2575 2500 # The first line of whatever came instead, flattened. Enough to tell a rate
2576 2501 # limit from a login page from a 404, and short enough to sit in a log line.
2577 2502 defp static_body_hint(body) when is_binary(body) do
@@ -2609,74 +2534,74 @@ defmodule RoomGtfs.Worker.Static do
2609 2534 case HTTPoison.get(cfg.config.url) do
2610 2535 {:ok, result} ->
2611 2536 bcast(state.id, :extracting, 2, 10)
2537 + import_old_zip(result.body, cfg, state.id)
2612 2538
2613 − case result.body
2614 − |> :zip.unzip([:memory]) do
2615 − {:ok, files} ->
2616 − files
2617 − |> Enum.each(fn {name, data} ->
2618 − as_csv =
2619 − data
2620 − |> String.split("\n")
2621 − |> Enum.map(&replace/1)
2622 − |> Enum.filter(fn x -> x != "" end)
2623 − |> XP.parse_stream()
2624 −
2625 − Logger.info({"datum", as_csv})
2626 −
2627 − case name do
2628 − 'agency.txt' ->
2629 − write_file(as_csv, :agency, state.id)
2630 − bcast(state.id, :agency, 3, 9)
2631 −
2632 − 'calendar.txt' ->
2633 − write_file(as_csv, :calendar, state.id)
2634 − bcast(state.id, :calendar, 4, 9)
2539 + {:error, info} ->
2540 + Logger.info(info.reason)
2541 + end
2635 2542
2636 − 'directions.txt' ->
2637 − write_file(as_csv, :directions, state.id)
2638 − bcast(state.id, :directions, 5, 9)
2543 + false ->
2544 + bcast(state.id, :disabled)
2545 + {:noreply, state}
2546 + end
2639 2547
2640 − 'routes.txt' ->
2641 − write_file(as_csv, :routes, state.id)
2642 − bcast(state.id, :routes, 6, 9)
2548 + {:noreply, state}
2549 + end
2643 2550
2644 − 'stops.txt' ->
2645 − write_file(as_csv, :stops, state.id)
2646 − bcast(state.id, :stops, 7, 9)
2551 + defp import_old_zip(body, cfg, id) do
2552 + case :zip.unzip(body, [:memory]) do
2553 + {:ok, files} ->
2554 + Enum.each(files, fn {name, data} -> import_old_file(name, data, id) end)
2647 2555
2648 − 'stop_times.txt' ->
2649 − write_file(data, :stop_times, state.id, via: :copy)
2650 − bcast(state.id, :stop_times, 8, 9)
2556 + Logger.info("GTFS::#{id} completed import")
2651 2557
2652 − 'trips.txt' ->
2653 − write_file(as_csv, :trips, state.id)
2654 − bcast(state.id, :trips, 9, 9)
2558 + Configuration.update_source_meta(cfg, %{last_run: DateTime.utc_now()})
2655 2559
2656 − _other ->
2657 − :ok
2658 − end
2659 − end)
2560 + {:error, _info} ->
2561 + bcast(id, :error, 1, 1)
2562 + Logger.info("GTFS::#{id} Got invalid zip file #{_info.reason}")
2563 + end
2564 + end
2660 2565
2661 − Logger.info("GTFS::#{state.id} completed import")
2566 + # What the zip-in-memory import above made of one file.
2567 + @old_files %{
2568 + 'agency.txt' => {:agency, 3},
2569 + 'calendar.txt' => {:calendar, 4},
2570 + 'directions.txt' => {:directions, 5},
2571 + 'routes.txt' => {:routes, 6},
2572 + 'stops.txt' => {:stops, 7},
2573 + 'trips.txt' => {:trips, 9}
2574 + }
2662 2575
2663 − Configuration.update_source_meta(cfg, %{last_run: DateTime.utc_now()})
2576 + defp import_old_file('stop_times.txt', data, id) do
2577 + log_old_datum(data)
2578 + write_file(data, :stop_times, id, via: :copy)
2579 + bcast(id, :stop_times, 8, 9)
2580 + end
2664 2581
2665 − {:error, _info} ->
2666 − bcast(state.id, :error, 1, 1)
2667 − Logger.info("GTFS::#{state.id} Got invalid zip file #{_info.reason}")
2668 − end
2582 + defp import_old_file(name, data, id) do
2583 + as_csv = log_old_datum(data)
2669 2584
2670 − {:error, info} ->
2671 − Logger.info(info.reason)
2672 − end
2585 + case Map.fetch(@old_files, name) do
2586 + {:ok, {type, step}} ->
2587 + write_file(as_csv, type, id)
2588 + bcast(id, type, step, 9)
2673 2589
2674 − false ->
2675 − bcast(state.id, :disabled)
2676 − {:noreply, state}
2590 + :error ->
2591 + :ok
2677 2592 end
2593 + end
2678 2594
2679 − {:noreply, state}
2595 + defp log_old_datum(data) do
2596 + as_csv =
2597 + data
2598 + |> String.split("\n")
2599 + |> Enum.map(&replace/1)
2600 + |> Enum.filter(fn x -> x != "" end)
2601 + |> XP.parse_stream()
2602 +
2603 + Logger.info({"datum", as_csv})
2604 + as_csv
2680 2605 end
2681 2606 end
2682 2607
@@ -2826,27 +2751,31 @@ defmodule RoomGtfs.Debug do
2826 2751 IO.puts("rt_tu is nil — no successful fetch yet")
2827 2752
2828 2753 feed ->
2829 − entities = feed.entity |> Enum.filter(& &1.trip_update)
2830 −
2831 − entities =
2832 − if stop_id do
2833 − entities
2834 − |> Enum.filter(fn e ->
2835 − Enum.any?(e.trip_update.stop_time_update, &(&1.stop_id == stop_id))
2836 − end)
2837 − else
2838 − entities
2839 − end
2840 −
2841 − IO.puts(
2842 − "#{length(entities)} trip_update entities#{if stop_id, do: " for stop #{stop_id}", else: ""}:"
2843 − )
2844 −
2845 − entities |> show(limit: :infinity)
2754 + dump_feed(feed, stop_id)
2846 2755 end
2847 2756 end
2848 2757 end
2849 2758
2759 + defp dump_feed(feed, stop_id) do
2760 + entities = feed.entity |> Enum.filter(& &1.trip_update)
2761 +
2762 + entities =
2763 + if stop_id do
2764 + entities
2765 + |> Enum.filter(fn e ->
2766 + Enum.any?(e.trip_update.stop_time_update, &(&1.stop_id == stop_id))
2767 + end)
2768 + else
2769 + entities
2770 + end
2771 +
2772 + IO.puts(
2773 + "#{length(entities)} trip_update entities#{if stop_id, do: " for stop #{stop_id}", else: ""}:"
2774 + )
2775 +
2776 + entities |> show(limit: :infinity)
2777 + end
2778 +
2850 2779 @doc """
2851 2780 Diagnose why live times aren't appearing for a stop.
2852 2781 Compares static trip IDs against RT feed trip IDs and stop_id formats,

Parents: f1dab56