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.

Place subway trains at their stop, and put realtime on the tint map

8b198ec · Gabriel Morell · 2026-08-24 05:57

13 files +269 -76
Message 
{commit_body(@commit)}

Files changed

modified apps/room_gtfs/lib/room_gtfs/gtfs_worker.ex
+97 −74
@@ -265,6 +265,9 @@ defmodule RoomGtfs.Worker.RT do
265 265 alias RoomSanctum.Repo
266 266 alias RoomGtfs.FeedCache
267 267
268 + # Static stops only change when a feed is reimported.
269 + @stops_ttl :timer.minutes(5)
270 +
268 271 def start_link(opts) do
269 272 GenServer.start_link(__MODULE__, opts, name: via_tuple("gtfs-rt" <> opts[:name]))
270 273 end
@@ -310,7 +313,11 @@ defmodule RoomGtfs.Worker.RT do
310 313 inst: nil,
311 314 rt_sa: nil,
312 315 rt_tu: nil,
313 − rt_vp: nil
316 + rt_vp: nil,
317 + # stop_id => {lat, lon}, for feeds that report a station instead of a
318 + # coordinate. See vehicle_positions_from/2.
319 + stops: %{},
320 + stops_at: nil
314 321 }}
315 322 end
316 323
@@ -451,31 +458,87 @@ defmodule RoomGtfs.Worker.RT do
451 458 defp feed_summary(_), do: %{loaded: false}
452 459
453 460 # Helper function to extract vehicle positions from protobuf data
454 − defp extract_vehicle_positions(data_vp) do
455 − case data_vp do
456 − nil ->
457 − []
458 −
459 − %{entity: entities} ->
460 − entities
461 − |> Enum.filter(fn entity -> entity.vehicle && entity.vehicle.position end)
462 − |> Enum.map(fn entity ->
463 − %{
464 − vehicle_id: entity.vehicle.vehicle.id,
465 − trip_id: if(entity.vehicle.trip, do: entity.vehicle.trip.trip_id, else: nil),
466 − route_id: if(entity.vehicle.trip, do: entity.vehicle.trip.route_id, else: nil),
467 − latitude: entity.vehicle.position.latitude,
468 − longitude: entity.vehicle.position.longitude,
469 − bearing: entity.vehicle.position.bearing,
470 − timestamp: entity.vehicle.timestamp
471 − }
472 − end)
473 −
461 + @doc """
462 + Vehicles from a realtime feed, placed on the map.
463 +
464 + A GTFS-RT VehiclePosition may carry a `position` and may not. Buses do:
465 + every one of the MTA's 318 had a latitude. NYCT subway trains do not -- not
466 + one of 45 did -- because what the subway reports is which station a train is
467 + at or heading to, `stop_id` plus `current_status`, and never a coordinate.
468 + Filtering on `position` therefore drew every bus and no train at all.
469 +
470 + So a train without a position is placed at the stop it names, and marked
471 + `position_inferred: true` so nothing downstream mistakes it for a fix. For a
472 + subway that is arguably the more useful reading anyway: you care which
473 + station it is at, not where it is in the tunnel.
474 +
475 + `stops` is a `stop_id => {lat, lon}` map from the source's own static feed.
476 + Empty, this behaves as it always did.
477 + """
478 + def vehicle_positions_from(nil, _stops), do: []
479 +
480 + def vehicle_positions_from(%{entity: entities}, stops) do
481 + entities
482 + |> Enum.filter(& &1.vehicle)
483 + |> Enum.flat_map(fn entity -> [place_vehicle(entity.vehicle, stops)] end)
484 + |> Enum.reject(&is_nil/1)
485 + end
486 +
487 + def vehicle_positions_from(_feed, _stops), do: []
488 +
489 + defp place_vehicle(v, stops) do
490 + base = %{
491 + vehicle_id: v.vehicle && v.vehicle.id,
492 + trip_id: v.trip && v.trip.trip_id,
493 + route_id: v.trip && v.trip.route_id,
494 + stop_id: v.stop_id,
495 + current_status: v.current_status,
496 + timestamp: v.timestamp
497 + }
498 +
499 + case {v.position, v.stop_id && Map.get(stops, v.stop_id)} do
500 + {%{latitude: lat, longitude: lon} = pos, _} when not is_nil(lat) and not is_nil(lon) ->
501 + Map.merge(base, %{
502 + latitude: lat,
503 + longitude: lon,
504 + bearing: pos.bearing,
505 + position_inferred: false
506 + })
507 +
508 + {_, {lat, lon}} when not is_nil(lat) and not is_nil(lon) ->
509 + # No bearing: the stop knows where it is, not which way the train faces.
510 + Map.merge(base, %{
511 + latitude: lat,
512 + longitude: lon,
513 + bearing: nil,
514 + position_inferred: true
515 + })
516 +
474 517 _ ->
475 − []
518 + nil
476 519 end
477 520 end
478 521
522 + # Static stops for this source, as the fallback above needs them. Reloaded at
523 + # most every few minutes: they only change when a feed is reimported, and a
524 + # subway feed is 1,488 rows that would otherwise be read on every poll.
525 + defp ensure_stops(state) do
526 + now = System.monotonic_time(:millisecond)
527 + age = now - Map.get(state, :stops_at, -@stops_ttl)
528 +
529 + if age >= @stops_ttl do
530 + stops =
531 + state.id
532 + |> Storage.list_stops()
533 + |> Map.new(fn stop -> {stop.stop_id, {stop.stop_lat, stop.stop_lon}} end)
534 +
535 + state |> Map.put(:stops, stops) |> Map.put(:stops_at, now)
536 + else
537 + state
538 + end
539 + end
540 +
541 +
479 542 def fetch_rt_url(url) do
480 543 case HTTPoison.get(url, [], follow_redirect: true) do
481 544 {:ok, %{status_code: 200} = result} ->
@@ -651,10 +714,10 @@ defmodule RoomGtfs.Worker.RT do
651 714 val ->
652 715 case FeedCache.get(val) do
653 716 {:ok, data_vp} when is_struct(data_vp, TransitRealtime.FeedMessage) ->
654 − new_state = state |> Map.put(:rt_vp, data_vp)
655 −
717 + new_state = state |> Map.put(:rt_vp, data_vp) |> ensure_stops()
718 +
656 719 # Broadcast vehicle position updates
657 − vehicles = extract_vehicle_positions(data_vp)
720 + vehicles = vehicle_positions_from(data_vp, new_state.stops)
658 721
659 722 # Broadcast to specific source channel for source page
660 723 Phoenix.PubSub.broadcast(
@@ -723,59 +786,19 @@ defmodule RoomGtfs.Worker.RT do
723 786 end
724 787 end
725 788
789 + # Both of these place vehicles through vehicle_positions_from/2, so a pull
790 + # here and the push broadcast on each poll agree about where a train is.
726 791 def handle_call(:query_vehicle_positions, _from, state) do
727 − # Return all vehicle positions
728 − case state.rt_vp do
729 − nil ->
730 − {:reply, [], state}
731 −
732 − _otherwise ->
733 − vehicles = state.rt_vp.entity
734 − |> Enum.filter(fn entity -> entity.vehicle && entity.vehicle.position end)
735 − |> Enum.map(fn entity ->
736 − %{
737 − vehicle_id: entity.vehicle.vehicle.id,
738 − trip_id: if(entity.vehicle.trip, do: entity.vehicle.trip.trip_id, else: nil),
739 − route_id: if(entity.vehicle.trip, do: entity.vehicle.trip.route_id, else: nil),
740 − latitude: entity.vehicle.position.latitude,
741 − longitude: entity.vehicle.position.longitude,
742 − bearing: entity.vehicle.position.bearing,
743 − timestamp: entity.vehicle.timestamp
744 − }
745 − end)
746 −
747 − {:reply, vehicles, state}
748 − end
792 + {:reply, vehicle_positions_from(state.rt_vp, state.stops), state}
749 793 end
750 794
751 795 def handle_call({:query_vehicle_positions, trips}, _from, state) do
752 − # Return vehicle positions for specific trips
753 − case state.rt_vp do
754 − nil ->
755 − {:reply, [], state}
756 −
757 − _otherwise ->
758 − vehicles = state.rt_vp.entity
759 − |> Enum.filter(fn entity ->
760 − entity.vehicle &&
761 − entity.vehicle.position &&
762 − entity.vehicle.trip &&
763 − Enum.member?(trips, entity.vehicle.trip.trip_id)
764 − end)
765 − |> Enum.map(fn entity ->
766 − %{
767 − vehicle_id: entity.vehicle.vehicle.id,
768 − trip_id: entity.vehicle.trip.trip_id,
769 − route_id: entity.vehicle.trip.route_id,
770 − latitude: entity.vehicle.position.latitude,
771 − longitude: entity.vehicle.position.longitude,
772 − bearing: entity.vehicle.position.bearing,
773 − timestamp: entity.vehicle.timestamp
774 − }
775 − end)
776 −
777 − {:reply, vehicles, state}
778 − end
796 + vehicles =
797 + state.rt_vp
798 + |> vehicle_positions_from(state.stops)
799 + |> Enum.filter(&(&1.trip_id && &1.trip_id in trips))
800 +
801 + {:reply, vehicles, state}
779 802 end
780 803 end
781 804
added apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/live_layers.ex
+102 −0
@@ -0,0 +1,102 @@
1 +defmodule RoomSanctumWeb.SourceLive.LiveLayers do
2 + @moduledoc """
3 + What a source has moving right now: vehicles, loose bikes, aircraft.
4 +
5 + The companion to `RoomSanctumWeb.SourceLive.Stations`, which answers the same
6 + question about places that do not move. Both exist because the tint map draws
7 + many sources at once and needs one shape per layer regardless of what kind of
8 + source produced it.
9 +
10 + Every read here goes through a worker rather than the database, and a worker
11 + may not be running -- a source added a minute ago, one whose supervisor is
12 + restarting, an app absent from this environment. All of those come back as an
13 + empty layer, because a map missing its buses is worth more than a page that
14 + will not render.
15 +
16 + `SourceLive.Show` gathers the same three things its own way, driven by PubSub
17 + for one source at a time. That is a better fit for a page watching a single
18 + feed, and merging the two is not obviously an improvement; what would be
19 + worth sharing is this module's habit of never letting a dead worker through.
20 + """
21 +
22 + alias RoomSanctum.Storage
23 +
24 + @doc """
25 + Everything live for one source, as `%{vehicles:, free_bikes:, aircraft:,
26 + route_types:}`.
27 +
28 + `route_types` comes back alongside the vehicles because a vehicle position
29 + names a route and the glyph is chosen from the route's type, so the caller
30 + would otherwise have to know to look it up separately.
31 + """
32 + def for_source(%{type: :gtfs, id: id}) do
33 + vehicles =
34 + safely(fn -> RoomGtfs.Worker.get_current_vehicle_positions(id) end, [])
35 + |> Storage.with_trip_context(id)
36 +
37 + %{
38 + vehicles: vehicles,
39 + free_bikes: [],
40 + aircraft: [],
41 + # Only worth the query if something is actually being drawn.
42 + route_types: if(vehicles == [], do: %{}, else: Storage.route_types(id))
43 + }
44 + end
45 +
46 + def for_source(%{type: :gbfs, id: id}) do
47 + bikes =
48 + safely(fn -> Storage.list_gbfs_free_bike_status() end, [])
49 + |> Enum.filter(&(&1.source_id == id))
50 +
51 + %{vehicles: [], free_bikes: bikes, aircraft: [], route_types: %{}}
52 + end
53 +
54 + def for_source(%{type: :icarus, id: id}) do
55 + %{vehicles: [], free_bikes: [], aircraft: aircraft(id), route_types: %{}}
56 + end
57 +
58 + def for_source(_source), do: empty()
59 +
60 + @doc "Merged layers across several sources, ready to hand to the map."
61 + def for_sources(sources) do
62 + sources
63 + |> Enum.map(&for_source/1)
64 + |> Enum.reduce(empty(), fn layer, acc ->
65 + %{
66 + vehicles: acc.vehicles ++ layer.vehicles,
67 + free_bikes: acc.free_bikes ++ layer.free_bikes,
68 + aircraft: acc.aircraft ++ layer.aircraft,
69 + # Route ids are namespaced per source in the static feed, so a plain
70 + # merge is safe: two agencies both having a route "1" would collide, but
71 + # route_types is keyed on the id the vehicle reports and both agencies'
72 + # vehicles report their own.
73 + route_types: Map.merge(acc.route_types, layer.route_types)
74 + }
75 + end)
76 + end
77 +
78 + @doc "Whether a source can contribute a live layer at all."
79 + def live?(%{type: type}), do: type in [:gtfs, :gbfs, :icarus]
80 + def live?(_source), do: false
81 +
82 + defp empty, do: %{vehicles: [], free_bikes: [], aircraft: [], route_types: %{}}
83 +
84 + # room_icarus is not a declared dependency of room_sanctum -- the umbrella
85 + # loads it at runtime, but it is absent from this app's test env, and an
86 + # unloaded module raises rather than exits.
87 + defp aircraft(id) do
88 + if Code.ensure_loaded?(RoomIcarus.Worker) do
89 + safely(fn -> RoomIcarus.Worker.current_aircraft(id) end, [])
90 + else
91 + []
92 + end
93 + end
94 +
95 + defp safely(fun, default) do
96 + fun.()
97 + rescue
98 + _error -> default
99 + catch
100 + :exit, _reason -> default
101 + end
102 +end
modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/map_view.ex
+48 −1
@@ -27,6 +27,7 @@ defmodule RoomSanctumWeb.SourceLive.MapView do
27 27 import RoomSanctumWeb.Components.QueryGeospatialMap
28 28
29 29 alias RoomSanctum.Configuration
30 + alias RoomSanctumWeb.SourceLive.LiveLayers
30 31 alias RoomSanctumWeb.SourceLive.Stations
31 32
32 33 # A backstop, not a budget. An offering's own map draws MBTA's ten thousand
@@ -38,8 +39,16 @@ defmodule RoomSanctumWeb.SourceLive.MapView do
38 39 # loud, rather than as a wedged browser.
39 40 @max_markers 50_000
40 41
42 + # Matches the source page. The realtime workers poll every 30s, so asking more
43 + # often than that returns the same feed.
44 + @refresh :timer.seconds(30)
45 +
41 46 @impl true
42 47 def mount(_params, _session, socket) do
48 + # Only once connected: the static render would pay for every worker read and
49 + # then be replaced.
50 + if connected?(socket), do: Process.send_after(self(), :refresh_live, 100)
51 +
43 52 {:ok,
44 53 socket
45 54 |> assign(:tint, nil)
@@ -47,7 +56,28 @@ defmodule RoomSanctumWeb.SourceLive.MapView do
47 56 |> assign(:stations, [])
48 57 |> assign(:station_statuses, [])
49 58 |> assign(:counts, %{})
50 − |> assign(:max_markers, @max_markers)}
59 + |> assign(:max_markers, @max_markers)
60 + |> assign(:show_aircraft, false)
61 + |> assign_live(%{vehicles: [], free_bikes: [], aircraft: [], route_types: %{}})}
62 + end
63 +
64 + @impl true
65 + def handle_info(:refresh_live, socket) do
66 + Process.send_after(self(), :refresh_live, @refresh)
67 + {:noreply, assign_live(socket, LiveLayers.for_sources(socket.assigns.sources))}
68 + end
69 +
70 + @impl true
71 + def handle_event("toggle-aircraft", _params, socket) do
72 + {:noreply, assign(socket, :show_aircraft, not socket.assigns.show_aircraft)}
73 + end
74 +
75 + defp assign_live(socket, layers) do
76 + socket
77 + |> assign(:vehicles, layers.vehicles)
78 + |> assign(:free_bikes, layers.free_bikes)
79 + |> assign(:aircraft, layers.aircraft)
80 + |> assign(:route_types, layers.route_types)
51 81 end
52 82
53 83 @impl true
@@ -89,6 +119,23 @@ defmodule RoomSanctumWeb.SourceLive.MapView do
89 119 |> assign(:stations, Enum.flat_map(per_source, fn {_source, stations} -> stations end))
90 120 |> assign(:counts, Map.new(per_source, fn {source, stations} -> {source.id, length(stations)} end))
91 121 |> assign(:station_statuses, Enum.flat_map(sources, &Stations.statuses_for_source/1))
122 + |> then(fn s ->
123 + # The sources just changed, so anything live already assigned describes the
124 + # tint that was on screen a moment ago.
125 + if connected?(s), do: assign_live(s, LiveLayers.for_sources(sources)), else: s
126 + end)
127 + end
128 +
129 + defp any_aircraft?(sources), do: Enum.any?(sources, &(&1.type == :icarus))
130 +
131 + # Subway trains report a station rather than a coordinate, so they are drawn at
132 + # the stop they name. Saying how many is the difference between a map that is
133 + # slightly approximate and one that is quietly lying.
134 + defp inferred_note(vehicles) do
135 + case Enum.count(vehicles, &Map.get(&1, :position_inferred)) do
136 + 0 -> ""
137 + n -> " (#{n} placed at a stop)"
138 + end
92 139 end
93 140
94 141 defp mappable?(source), do: Stations.mappable?(source)
modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/map_view.html.heex
+22 −1
@@ -35,7 +35,7 @@
35 35 <% end %>
36 36 </div>
37 37
38 − <%= if @stations == [] do %>
38 + <%= if @stations == [] and @vehicles == [] and @free_bikes == [] and @aircraft == [] do %>
39 39 <div class="card card-compact w-full bg-base-100 shadow-xl mx-6 my-2">
40 40 <div class="card-body">
41 41 <h3 class="card-title italic">Nothing to draw</h3>
@@ -58,12 +58,33 @@
58 58 queries={[]}
59 59 stations={@stations}
60 60 station_statuses={@station_statuses}
61 + vehicle_positions={@vehicles}
62 + route_types={@route_types}
63 + free_bikes={@free_bikes}
64 + aircraft={if @show_aircraft, do: @aircraft, else: []}
65 + aircraft_event={if any_aircraft?(@sources), do: "toggle-aircraft"}
66 + show_aircraft={@show_aircraft}
61 67 source_tint={@tint}
62 68 selected_tint={@tint}
63 69 max_markers={@max_markers}
64 70 class=""
65 71 height="700px"
66 72 />
73 +
74 + <%!-- Counts rather than a legend entry: the map's own legend explains the
75 + marker kinds, and what is useful here is knowing whether a layer is
76 + empty because nothing is running or because no source supplies it. --%>
77 + <div class="mt-2 text-xs opacity-70 flex flex-wrap gap-3">
78 + <span><%= length(@stations) %> places</span>
79 + <span :if={@vehicles != []}>
80 + <%= length(@vehicles) %> vehicles<%= inferred_note(@vehicles) %>
81 + </span>
82 + <span :if={@free_bikes != []}><%= length(@free_bikes) %> loose bikes</span>
83 + <span :if={@aircraft != [] and @show_aircraft}><%= length(@aircraft) %> aircraft</span>
84 + <span :if={Enum.any?(@sources, &LiveLayers.live?(&1)) and @vehicles == [] and @free_bikes == []}>
85 + nothing moving right now
86 + </span>
87 + </div>
67 88 </div>
68 89 <% end %>
69 90 <% end %>

Parents: 011a33a