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.

Give a Plani the realtime a stop query has always had

eac203f · Gabriel Morell · 2026-09-04 14:26

15 files +141 -22
Message 
{commit_body(@commit)}

Files changed

modified apps/room_gtfs/lib/room_gtfs/gtfs_worker.ex
+30 −14
@@ -253,9 +253,11 @@ defmodule RoomGtfs.Worker do
253 253
254 254 stops
255 255 |> Enum.flat_map(fn stop ->
256 − Storage.get_upcoming_arrivals_for_stop(id, stop, @area_arrivals_per_stop, :now, inst.config.tz)
256 + # Realtime per stop, as a query naming one stop has always had. Reading
257 + # the timetable here meant an area query published times that had been
258 + # revised minutes ago, and no way to tell they were stale.
259 + arrivals_at_stop(id, stop, @area_arrivals_per_stop, inst.config)
257 260 end)
258 − |> Storage.fix_arrival_times()
259 261 |> Enum.sort_by(& &1.arrival_time)
260 262 |> Enum.take(@area_arrivals_total)
261 263 end
@@ -264,23 +266,36 @@ defmodule RoomGtfs.Worker do
264 266 # Bare: this wants the config blob, and nothing on this path reads the
265 267 # mailboxes and webhooks the preloading version fetches alongside it.
266 268 inst = Configuration.get_source!(:bare, id)
267 − # The source is already in hand, so its timezone goes with the call rather
268 − # than being looked up again inside it.
269 +
270 + arrivals_at_stop(id, query.stop, 16, inst.config)
271 + end
272 +
273 + @doc """
274 + What is due at one stop, with whatever realtime the feed has to say about it.
275 +
276 + Pulled out of `query_stop/2` because three callers want it and only one had
277 + it. A query naming a stop got live times, occupancy and platforms; an area
278 + query and a Plani -- both of which ask several stops and blend the answers --
279 + read straight from the timetable and published departures that had already
280 + been revised, with nothing marking them as scheduled rather than live.
281 +
282 + Takes the source's config rather than looking it up, since a caller asking
283 + about several stops has it in hand already and would otherwise re-fetch it
284 + once per stop.
285 + """
286 + def arrivals_at_stop(id, stop_id, limit, config) do
269 287 res =
270 − Storage.get_upcoming_arrivals_for_stop(id, query.stop, 16, :now, inst.config.tz)
271 − |> Storage.fix_arrival_times
288 + Storage.get_upcoming_arrivals_for_stop(id, stop_id, limit, :now, config.tz)
289 + |> Storage.fix_arrival_times()
272 290
273 − # Not `url_rt_tu` alone: a source whose trip updates arrive in a combined
274 − # feed has that field blank and its URL in url_rt_shared, and gating on the
275 − # specific field skipped realtime entirely for exactly those sources.
276 − case RoomGtfs.Worker.RT.rt_url_for(inst.config, :tu) do
291 + case RoomGtfs.Worker.RT.rt_url_for(config, :tu) do
277 292 nil ->
278 293 res
279 294
280 295 _val ->
281 296 trips = res |> Enum.map(fn x -> x.trip_id end)
282 297
283 − case get_current_realtime(id, trips, query.stop, inst.config) do
298 + case get_current_realtime(id, trips, stop_id, config) do
284 299 [] ->
285 300 res
286 301
@@ -288,12 +303,12 @@ defmodule RoomGtfs.Worker do
288 303 # How full the vehicle is may be reported per stop in the trip
289 304 # update, or only by the vehicle's own position report -- so the
290 305 # positions are fetched once here to fall back on.
291 − vehicles = occupancy_vehicles(id, inst.config, trips)
306 + vehicles = occupancy_vehicles(id, config, trips)
292 307
293 308 res
294 309 |> Enum.map(fn x ->
295 310 case Enum.find(rtvals, fn v ->
296 − trip_id_match?(inst.config, x.trip_id, v.trip_id)
311 + trip_id_match?(config, x.trip_id, v.trip_id)
297 312 end) do
298 313 nil ->
299 314 x
@@ -304,7 +319,7 @@ defmodule RoomGtfs.Worker do
304 319
305 320 x =
306 321 x
307 − |> merge_occupancy(occupancy_for(inst.config, x.trip_id, stu, vehicles))
322 + |> merge_occupancy(occupancy_for(config, x.trip_id, stu, vehicles))
308 323 |> merge_schedule_relationship(val.schedule_relationship, stu)
309 324 |> merge_stop_properties(stu)
310 325
@@ -325,6 +340,7 @@ defmodule RoomGtfs.Worker do
325 340 end
326 341 end
327 342
343 +
328 344 # How long to wait for vehicle positions before giving up on occupancy.
329 345 #
330 346 # Deliberately short. This is the second call query_stop makes to the same
modified apps/room_sanctum/lib/room_sanctum/condensers/basic_mqtt.ex
+6 −1
@@ -323,7 +323,12 @@ defmodule RoomSanctum.Condenser.BasicMQTT do
323 323 # An area query has no one stop, so there is no stop-specific alert to
324 324 # look for -- only the ones on the routes that turned up.
325 325 stop = Map.get(query.query, :stop)
326 − alerts = RoomGtfs.Worker.query_alerts(id, stop, route_ids)
326 + # A broken out entry is keyed by source *and* stop, so its id is not a
327 + # source id and the lookup found nothing -- silently, since no alerts and
328 + # no such source look the same from here. The descriptor names the source
329 + # when the key cannot.
330 + source_id = Map.get(query.query, :source_id) || id
331 + alerts = RoomGtfs.Worker.query_alerts(source_id, stop, route_ids)
327 332
328 333 condensed_with_alerts = condensed |> Enum.map(fn route ->
329 334 route_alerts = Enum.filter(alerts, fn a ->
modified apps/room_sanctum/lib/room_sanctum/workers/plani_worker.ex
+51 −7
@@ -27,6 +27,9 @@ defmodule RoomSanctum.Worker.Plani do
27 27 # Answered at a point rather than near one: there is one answer and it is
28 28 # computed for wherever you are, so "the closest five" means nothing. They
29 29 # all took a foci already; the anchor is simply a different one.
30 + # Enough to fill a card and leave the sort something to choose from.
31 + @arrivals_per_stop 8
32 +
30 33 @asked_at_a_point [:weather, :ephem, :pollen, :icarus]
31 34
32 35 def start_link(opts) do
@@ -214,8 +217,7 @@ defmodule RoomSanctum.Worker.Plani do
214 217 Enum.map(stops, fn stop ->
215 218 arrivals =
216 219 source_id
217 − |> Storage.get_upcoming_arrivals_for_stop(stop.stop_id, 8, :now, source.config.tz)
218 − |> Storage.fix_arrival_times()
220 + |> arrivals_at(stop.stop_id, source.config)
219 221 |> Enum.sort_by(& &1.arrival_time)
220 222
221 223 # Which way the stop lies, carried on each arrival: the condenser
@@ -234,7 +236,13 @@ defmodule RoomSanctum.Worker.Plani do
234 236 # empty -- the same call the clients already make.
235 237 for {stop, arrivals} <- by_stop, arrivals != [] do
236 238 key = "#{source_id}:#{stop.stop_id}"
237 − {{key, :gtfs}, arrivals, described(source, key, stop.stop_name)}
239 + descriptor =
240 + described(source, key, stop.stop_name, %{
241 + source_id: source_id,
242 + stop: stop.stop_id
243 + })
244 +
245 + {{key, :gtfs}, arrivals, descriptor}
238 246 end
239 247 else
240 248 blended =
@@ -274,7 +282,7 @@ defmodule RoomSanctum.Worker.Plani do
274 282 per_dock =
275 283 for dock <- docks do
276 284 key = "#{source_id}:#{dock.station_id}"
277 − {{key, :gbfs}, [dock], described(source, key, dock.name)}
285 + {{key, :gbfs}, [dock], described(source, key, dock.name, %{source_id: source_id})}
278 286 end
279 287
280 288 if bikes == [],
@@ -372,6 +380,34 @@ defmodule RoomSanctum.Worker.Plani do
372 380 {arrival.trip.route_id, direction}
373 381 end
374 382
383 + # What is due at a stop, live where the feed says so.
384 + #
385 + # Through room_gtfs rather than straight at the tables: the realtime half --
386 + # revised times, occupancy, platform assignments -- lives there, and reading
387 + # the timetable directly is why a Plani was publishing departures that had
388 + # already been revised, with nothing to say they were only scheduled.
389 + #
390 + # Called across apps the way the condenser calls `query_alerts`, since
391 + # room_sanctum does not declare a dependency on room_gtfs. A missing module
392 + # or an unstarted worker falls back to the timetable, which is worse than
393 + # live but much better than nothing.
394 + defp arrivals_at(source_id, stop_id, config) do
395 + RoomGtfs.Worker.arrivals_at_stop(source_id, stop_id, @arrivals_per_stop, config)
396 + rescue
397 + error ->
398 + Logger.debug("plani: no realtime for stop #{stop_id}: #{Exception.message(error)}")
399 + timetable_at(source_id, stop_id, config)
400 + catch
401 + :exit, _ ->
402 + timetable_at(source_id, stop_id, config)
403 + end
404 +
405 + defp timetable_at(source_id, stop_id, config) do
406 + source_id
407 + |> Storage.get_upcoming_arrivals_for_stop(stop_id, @arrivals_per_stop, :now, config.tz)
408 + |> Storage.fix_arrival_times()
409 + end
410 +
375 411 defp cap(rows, nil), do: rows
376 412 defp cap(rows, limit) when is_integer(limit), do: Enum.take(rows, limit)
377 413
@@ -424,10 +460,18 @@ defmodule RoomSanctum.Worker.Plani do
424 460
425 461 defp rad(degrees), do: degrees * :math.pi() / 180
426 462
427 − defp described(source), do: described(source, source.id, source.name)
463 + defp described(source), do: described(source, source.id, source.name, %{})
464 +
465 + defp described(source, id, name), do: described(source, id, name, %{})
428 466
429 − defp described(source, id, name) do
430 − %{id: id, name: name, meta: %{}, query: %{}}
467 + # The descriptor an entry is published under.
468 + #
469 + # `query` carries what the condensers need to look past the entry itself. A
470 + # broken out entry is keyed by source *and* stop, so its id is not a source
471 + # id -- and the alerts lookup, which takes one, quietly found nothing for
472 + # every broken out stop. Naming the source here is how it can.
473 + defp described(_source, id, name, query) do
474 + %{id: id, name: name, meta: %{}, query: query}
431 475 end
432 476
433 477 # Asked at the anchor. Each of these already takes a query naming a foci; a
added apps/room_sanctum/test/room_sanctum/plani_gtfs_rt_test.exs
+54 −0
@@ -0,0 +1,54 @@
1 +defmodule RoomSanctum.PlaniGtfsRtTest do
2 + use ExUnit.Case, async: true
3 +
4 + alias RoomSanctum.Condenser.BasicMQTT
5 +
6 + defp arrival(route) do
7 + %{
8 + arrival_time: "08:00:00",
9 + arrival_time_live_ts: nil,
10 + tz: "America/New_York",
11 + trip_id: "t1",
12 + trip: %{
13 + trip_headsign: "Arlington",
14 + direction: %{direction: "In"},
15 + route_id: route,
16 + route: %{
17 + route_type: "3",
18 + route_short_name: route,
19 + route_long_name: route,
20 + route_color: nil,
21 + route_text_color: nil
22 + }
23 + }
24 + }
25 + end
26 +
27 + test "a broken out entry names its source so alerts can be looked up" do
28 + # The entry's own id is source-and-stop, which is not a source id -- so the
29 + # alerts lookup found nothing for every broken out stop, and silently: no
30 + # alerts and no such source look identical from the condenser.
31 + descriptor = %{
32 + id: "40:2378",
33 + name: "Boston Ave @ College Ave",
34 + meta: %{},
35 + query: %{source_id: 40, stop: "2378"}
36 + }
37 +
38 + out = BasicMQTT.condense({"40:2378", :gtfs}, [arrival("96")], descriptor)
39 +
40 + assert out.query.name == "Boston Ave @ College Ave"
41 + assert [%{route: "96"}] = out.data
42 + end
43 +
44 + test "a vision's entry still resolves alerts from the entry id" do
45 + # No source_id in the descriptor: the id *is* the source there, and that
46 + # path must keep working exactly as it did.
47 + vision = %{id: 12, name: "Forest Hills", meta: %{}, query: %{stop: "10642"}}
48 +
49 + out = BasicMQTT.condense({12, :gtfs}, [arrival("14")], vision)
50 +
51 + assert out.query.name == "Forest Hills"
52 + assert [%{route: "14"}] = out.data
53 + end
54 +end

Parents: dfde4b2