neiam /apollos_crib

It's Apollo's Crib

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

Import shapes.txt and linked_datasets.txt, and fix CR CR LF feeds

521323d · Gabriel Morell · 2026-08-10 22:10

23 files +752 -5

Message

{commit_body(@commit)}

Files changed

modified apps/room_gtfs/lib/room_gtfs/gtfs_worker.ex

+135 −5

@@ -701,10 +701,50 @@ end

701	701	\|> String.trim()
702	702	end
703	703
	704	+ # Postgres picks its newline style from the first terminator it sees. Given
	705	+ # CR CR LF -- which Caltrain's trips.txt ships -- it decides the file is
	706	+ # CR-terminated and then reads every second line as empty, failing with
	707	+ # "missing data for column". Collapsing the CR run before each LF fixes that,
	708	+ # and mapping a lone CR to LF keeps classic CR-only files loading.
	709	+ defp rewrite_newlines(binary) do
	710	+ binary
	711	+ \|> String.replace(~r/\r+\n/, "\n")
	712	+ \|> String.replace("\r", "\n")
	713	+ end
	714	+
	715	+ # A CR run at the end of a chunk may be finished by a LF at the start of the
	716	+ # next one, so it is held back rather than converted early.
	717	+ defp hold_trailing_cr(binary) do
	718	+ case Regex.run(~r/\r+\z/, binary, return: :index) do
	719	+ [{start, len}] -> {binary_part(binary, 0, start), binary_part(binary, start, len)}
	720	+ nil -> {binary, ""}
	721	+ end
	722	+ end
	723	+
	724	+ @doc false
	725	+ # Exposed only so the normaliser can be exercised directly; the importer
	726	+ # calls the private function.
	727	+ def normalize_newlines_for_test(contents), do: normalize_newlines(contents)
	728	+
	729	+ defp normalize_newlines(contents) do
	730	+ Stream.transform(
	731	+ contents,
	732	+ fn -> "" end,
	733	+ fn chunk, carry ->
	734	+ {body, pending} = hold_trailing_cr(carry <> IO.iodata_to_binary(chunk))
	735	+ {[rewrite_newlines(body)], pending}
	736	+ end,
	737	+ fn carry -> {[rewrite_newlines(carry)], ""} end,
	738	+ fn _carry -> :ok end
	739	+ )
	740	+ end
	741	+
704	742	defp write_file(contents, type, id, pid) do
705	743	datetime = NaiveDateTime.local_now()
706	744	Logger.info("GTFS::#{id} writing #{type} (c)")
707	745
	746	+ contents = normalize_newlines(contents)
	747	+
708	748	cols_j =
709	749	contents
710	750	\|> Stream.chunk_every(500)

@@ -732,6 +772,7 @@ end

732	772	:stops -> RoomSanctum.Storage.truncate_stop(id)
733	773	:stop_times -> RoomSanctum.Storage.truncate_stop_time(id)
734	774	:trips -> RoomSanctum.Storage.truncate_trip(id)
	775	+ :shapes -> RoomSanctum.Storage.truncate_shape(id)
735	776	_ -> :ok
736	777	end
737	778

@@ -745,6 +786,7 @@ end

745	786	:stops -> {:gtfs_stops, [GTFS.Stop \|> get_cols(cols_j_plus)], GTFS.Stop \|> get_cols_pgtypes(cols_j_plus)}
746	787	:stop_times -> {:gtfs_stop_times, [GTFS.StopTime \|> get_cols(cols_j_plus)], GTFS.StopTime \|> get_cols_pgtypes(cols_j_plus)}
747	788	:trips -> {:gtfs_trips, [GTFS.Trip \|> get_cols(cols_j_plus)], GTFS.Trip \|> get_cols_pgtypes(cols_j_plus)}
	789	+ :shapes -> {:gtfs_shapes, [GTFS.Shape \|> get_cols(cols_j_plus)], GTFS.Shape \|> get_cols_pgtypes(cols_j_plus)}
748	790	end
749	791
750	792	# IO.inspect({table, columns, pg_cols})

@@ -937,6 +979,76 @@ end

937	979	DateTime.utc_now()
938	980	end
939	981
	982	+
	983	+ # linked_datasets.txt associates GTFS-RT feeds with the schedule that
	984	+ # describes them, so a feed that ships one can wire up its own realtime URLs.
	985	+ #
	986	+ # Publishers disagree on the details, so this reads by header name rather
	987	+ # than position: MBTA ships five columns and writes the flags as 1/0, while
	988	+ # Caltrain ships seven and writes true/false.
	989	+ @rt_flag_fields [
	990	+ {"trip_updates", :url_rt_tu},
	991	+ {"vehicle_positions", :url_rt_vp},
	992	+ {"service_alerts", :url_rt_sa}
	993	+ ]
	994	+
	995	+ def parse_linked_datasets(csv) do
	996	+ case String.split(csv, ~r/\r?\n/, trim: true) do
	997	+ [] ->
	998	+ []
	999	+
	1000	+ [header \| rows] ->
	1001	+ keys = header \|> String.split(",") \|> Enum.map(&normalize_header/1)
	1002	+
	1003	+ Enum.map(rows, fn row ->
	1004	+ keys
	1005	+ \|> Enum.zip(String.split(row, ","))
	1006	+ \|> Map.new(fn {k, v} -> {k, String.trim(v)} end)
	1007	+ end)
	1008	+ end
	1009	+ end
	1010	+
	1011	+ defp rt_flag?(value) do
	1012	+ String.downcase(String.trim(value \|\| "")) in ["1", "true", "yes"]
	1013	+ end
	1014	+
	1015	+ # A feed behind a key cannot be fetched with what we have, so it is left for
	1016	+ # the user to fill in by hand rather than saved as a URL that always 401s.
	1017	+ defp rt_open?(row) do
	1018	+ case Map.get(row, "authentication_type") do
	1019	+ nil -> true
	1020	+ value -> String.downcase(String.trim(value)) in ["", "0", "none"]
	1021	+ end
	1022	+ end
	1023	+
	1024	+ def linked_dataset_urls(rows) do
	1025	+ Enum.reduce(@rt_flag_fields, %{}, fn {flag, field}, acc ->
	1026	+ row = Enum.find(rows, fn r -> rt_flag?(Map.get(r, flag)) and rt_open?(r) end)
	1027	+ url = row && String.trim(Map.get(row, "url") \|\| "")
	1028	+
	1029	+ if url in [nil, ""], do: acc, else: Map.put(acc, field, url)
	1030	+ end)
	1031	+ end
	1032	+
	1033	+ # Only ever fills blanks: a URL already in the config was put there
	1034	+ # deliberately and outranks whatever the feed advertises.
	1035	+ defp apply_linked_datasets(cfg, urls) do
	1036	+ fill =
	1037	+ urls
	1038	+ \|> Enum.reject(fn {field, _url} ->
	1039	+ existing = Map.get(cfg.config, field)
	1040	+ is_binary(existing) and String.trim(existing) != ""
	1041	+ end)
	1042	+ \|> Map.new()
	1043	+
	1044	+ if fill == %{} do
	1045	+ :ok
	1046	+ else
	1047	+ Logger.info("GTFS::#{cfg.id} linked_datasets supplied #{inspect(Map.keys(fill))}")
	1048	+ Configuration.update_source_config(cfg, fill)
	1049	+ end
	1050	+ end
	1051	+
940	1052	defp file_to_atom(filename) do
941	1053	case filename do
942	1054	"agency.txt" ->

@@ -959,6 +1071,9 @@ end

959	1071
960	1072	"trips.txt" ->
961	1073	:trips
	1074	+
	1075	+ "shapes.txt" ->
	1076	+ :shapes
962	1077	end
963	1078	end
964	1079

@@ -971,6 +1086,7 @@ end

971	1086	"stops.txt" -> 7
972	1087	"stop_times.txt" -> 8
973	1088	"trips.txt" -> 9
	1089	+ "shapes.txt" -> 10
974	1090	end
975	1091	end
976	1092

@@ -985,16 +1101,29 @@ end

985	1101	case cfg.enabled do
986	1102	true ->
987	1103	Logger.info("GTFS::#{state.id} updating static info")
988		− bcast(state.id, :downloading, 1, 9)
	1104	+ bcast(state.id, :downloading, 1, 10)
989	1105
990	1106	case HTTPoison.get(cfg.config.url, [], follow_redirect: true) do
991	1107	{:ok, result} ->
992		− bcast(state.id, :extracting, 2, 9)
	1108	+ bcast(state.id, :extracting, 2, 10)
993	1109
994	1110	case result.body \|> Unzip.InMem.new() \|> Unzip.new() do
995	1111	{:ok, unzip} ->
996	1112	files = Unzip.list_entries(unzip)
997	1113
	1114	+ case Enum.find(files, &(&1.file_name == "linked_datasets.txt")) do
	1115	+ nil ->
	1116	+ :ok
	1117	+
	1118	+ entry ->
	1119	+ Unzip.file_stream!(unzip, entry.file_name)
	1120	+ \|> Enum.to_list()
	1121	+ \|> IO.iodata_to_binary()
	1122	+ \|> parse_linked_datasets()
	1123	+ \|> linked_dataset_urls()
	1124	+ \|> then(&apply_linked_datasets(cfg, &1))
	1125	+ end
	1126	+
998	1127	try do
999	1128	files
1000	1129	\|> Enum.map(fn e ->

@@ -1007,11 +1136,12 @@ end

1007	1136	"stops.txt",
1008	1137	"stop_times.txt",
1009	1138	"trips.txt",
	1139	+ "shapes.txt",
1010	1140	],
1011	1141	e.file_name
1012	1142	) do
1013	1143
1014		− bcast(state.id, file_to_atom(e.file_name), file_to_order(e.file_name), 9)
	1144	+ bcast(state.id, file_to_atom(e.file_name), file_to_order(e.file_name), 10)
1015	1145
1016	1146	Unzip.file_stream!(unzip, e.file_name)
1017	1147	\|> write_file(file_to_atom(e.file_name), state.id, nil)

@@ -1046,11 +1176,11 @@ end

1046	1176	case cfg.enabled do
1047	1177	true ->
1048	1178	Logger.info("GTFS::#{state.id} updating static info")
1049		− bcast(state.id, :downloading, 1, 9)
	1179	+ bcast(state.id, :downloading, 1, 10)
1050	1180
1051	1181	case HTTPoison.get(cfg.config.url) do
1052	1182	{:ok, result} ->
1053		− bcast(state.id, :extracting, 2, 9)
	1183	+ bcast(state.id, :extracting, 2, 10)
1054	1184
1055	1185	case result.body
1056	1186	\|> :zip.unzip([:memory]) do

added apps/room_gtfs/test/linked_datasets_test.exs

+58 −0

@@ -0,0 +1,58 @@

	1	+defmodule RoomGtfs.LinkedDatasetsTest do
	2	+ @moduledoc """
	3	+ linked_datasets.txt wires a GTFS feed to its own GTFS-RT endpoints. Publishers
	4	+ disagree on the spelling, so these are the shapes seen in the wild.
	5	+ """
	6	+ use ExUnit.Case, async: true
	7	+
	8	+ describe "linked_datasets.txt" do
	9	+ test "reads MBTA's 1/0 flags across one row per feed" do
	10	+ csv = """
	11	+ url,trip_updates,vehicle_positions,service_alerts,authentication_type
	12	+ https://cdn.mbta.com/realtime/TripUpdates.pb,1,0,0,0
	13	+ https://cdn.mbta.com/realtime/VehiclePositions.pb,0,1,0,0
	14	+ https://cdn.mbta.com/realtime/Alerts.pb,0,0,1,0
	15	+ """
	16	+
	17	+ urls = csv \|> RoomGtfs.Worker.Static.parse_linked_datasets() \|> RoomGtfs.Worker.Static.linked_dataset_urls()
	18	+
	19	+ assert urls == %{
	20	+ url_rt_tu: "https://cdn.mbta.com/realtime/TripUpdates.pb",
	21	+ url_rt_vp: "https://cdn.mbta.com/realtime/VehiclePositions.pb",
	22	+ url_rt_sa: "https://cdn.mbta.com/realtime/Alerts.pb"
	23	+ }
	24	+ end
	25	+
	26	+ test "reads Caltrain's true/false flags and extra columns" do
	27	+ csv = """
	28	+ url,trip_updates,vehicle_positions,service_alerts,authentication_type,authentication_info_url,api_key_parameter_name
	29	+ http://example.test/service_alerts.proto,false,false,true,none,,
	30	+ """
	31	+
	32	+ urls = csv \|> RoomGtfs.Worker.Static.parse_linked_datasets() \|> RoomGtfs.Worker.Static.linked_dataset_urls()
	33	+
	34	+ assert urls == %{url_rt_sa: "http://example.test/service_alerts.proto"}
	35	+ end
	36	+
	37	+ test "a feed behind an api key is left alone" do
	38	+ csv = """
	39	+ url,trip_updates,vehicle_positions,service_alerts,authentication_type
	40	+ https://example.test/tu.pb,1,0,0,2
	41	+ https://example.test/vp.pb,0,1,0,0
	42	+ """
	43	+
	44	+ urls = csv \|> RoomGtfs.Worker.Static.parse_linked_datasets() \|> RoomGtfs.Worker.Static.linked_dataset_urls()
	45	+
	46	+ assert urls == %{url_rt_vp: "https://example.test/vp.pb"}
	47	+ refute Map.has_key?(urls, :url_rt_tu)
	48	+ end
	49	+
	50	+ test "a header-only or empty file yields nothing" do
	51	+ assert RoomGtfs.Worker.Static.parse_linked_datasets("") \|> RoomGtfs.Worker.Static.linked_dataset_urls() == %{}
	52	+
	53	+ assert "url,trip_updates,vehicle_positions,service_alerts\n"
	54	+ \|> RoomGtfs.Worker.Static.parse_linked_datasets()
	55	+ \|> RoomGtfs.Worker.Static.linked_dataset_urls() == %{}
	56	+ end
	57	+ end
	58	+end

added apps/room_gtfs/test/newline_normalisation_test.exs

+65 −0

@@ -0,0 +1,65 @@

	1	+defmodule RoomGtfs.NewlineNormalisationTest do
	2	+ @moduledoc """
	3	+ Feeds disagree about line endings, and COPY is unforgiving about it: Postgres
	4	+ infers the newline style from the first terminator, so CR CR LF (Caltrain's
	5	+ trips.txt) reads as CR-terminated and every second line comes out empty.
	6	+
	7	+ The normaliser is private, so these drive it the way the importer does --
	8	+ through a stream of chunks -- via the public entry point used in tests.
	9	+ """
	10	+ use ExUnit.Case, async: true
	11	+
	12	+ alias RoomGtfs.Worker.Static
	13	+
	14	+ defp normalize(chunks) do
	15	+ chunks
	16	+ \|> Static.normalize_newlines_for_test()
	17	+ \|> Enum.join()
	18	+ end
	19	+
	20	+ test "leaves plain LF alone" do
	21	+ assert normalize(["a,b\n1,2\n"]) == "a,b\n1,2\n"
	22	+ end
	23	+
	24	+ test "converts CRLF to LF" do
	25	+ assert normalize(["a,b\r\n1,2\r\n"]) == "a,b\n1,2\n"
	26	+ end
	27	+
	28	+ test "collapses the CR CR LF that breaks the import" do
	29	+ assert normalize(["a,b\r\r\n1,2\r\r\n"]) == "a,b\n1,2\n"
	30	+ end
	31	+
	32	+ test "collapses arbitrarily long CR runs" do
	33	+ assert normalize(["a,b\r\r\r\r\n1,2\n"]) == "a,b\n1,2\n"
	34	+ end
	35	+
	36	+ test "treats a lone CR as a terminator so CR-only files still load" do
	37	+ assert normalize(["a,b\r1,2\r"]) == "a,b\n1,2\n"
	38	+ end
	39	+
	40	+ test "a CR run split across chunks is still collapsed" do
	41	+ # the boundary case: the CRs end one chunk, the LF starts the next
	42	+ assert normalize(["a,b\r", "\r\n1,2\n"]) == "a,b\n1,2\n"
	43	+ assert normalize(["a,b\r\r", "\n1,2\n"]) == "a,b\n1,2\n"
	44	+ assert normalize(["a,b", "\r\r\n", "1,2\n"]) == "a,b\n1,2\n"
	45	+ end
	46	+
	47	+ test "a trailing CR at end of input is not swallowed" do
	48	+ assert normalize(["a,b\n1,2\r"]) == "a,b\n1,2\n"
	49	+ end
	50	+
	51	+ test "does not invent or drop records" do
	52	+ caltrain = "route_id,service_id,trip_id\r\r\n77119,c_717,163\r\r\n77119,c_717,167\r\r\n"
	53	+
	54	+ lines = caltrain \|> List.wrap() \|> normalize() \|> String.split("\n", trim: true)
	55	+
	56	+ assert length(lines) == 3
	57	+ assert hd(lines) == "route_id,service_id,trip_id"
	58	+ assert List.last(lines) == "77119,c_717,167"
	59	+ end
	60	+
	61	+ test "content is otherwise untouched" do
	62	+ row = ~s(1,"Main St, Apt 2",x\r\r\n)
	63	+ assert normalize([row]) == ~s(1,"Main St, Apt 2",x\n)
	64	+ end
	65	+end

modified apps/room_sanctum/lib/room_sanctum/storage.ex

+174 −0

@@ -358,6 +358,7 @@ defmodule RoomSanctum.Storage do

358	358	end
359	359
360	360	alias RoomSanctum.Storage.GTFS.Route
	361	+ alias RoomSanctum.Storage.GTFS.Shape
361	362
362	363	@doc """
363	364	Returns the list of routes.

@@ -377,6 +378,169 @@ defmodule RoomSanctum.Storage do

377	378	\|> Repo.all()
378	379	end
379	380
	381	+ @doc """
	382	+ The routes that call at a stop, as a list of route_ids.
	383	+
	384	+ Backed by the (source_id, stop_id) index on stop_times -- without it the
	385	+ planner walks every trip in the feed.
	386	+ """
	387	+ def routes_serving_stop(source_id, stop_id) do
	388	+ Repo.query!(
	389	+ """
	390	+ SELECT DISTINCT t.route_id
	391	+ FROM gtfs_stop_times st
	392	+ JOIN gtfs_trips t ON t.source_id = st.source_id AND t.trip_id = st.trip_id
	393	+ WHERE st.source_id = $1 AND st.stop_id = $2
	394	+ """,
	395	+ [source_id, stop_id]
	396	+ ).rows
	397	+ \|> Enum.map(&List.first/1)
	398	+ end
	399	+
	400	+ @doc """
	401	+ route_id => route_type for one source.
	402	+
	403	+ Realtime vehicle positions carry only a route_id, but what a vehicle is --
	404	+ bus, tram, ferry -- lives on the route, so callers resolve it once and carry
	405	+ the map rather than querying per vehicle.
	406	+ """
	407	+ def route_types(source_id) do
	408	+ from(r in Route,
	409	+ where: r.source_id == ^source_id,
	410	+ select: {r.route_id, r.route_type}
	411	+ )
	412	+ \|> Repo.all()
	413	+ \|> Map.new()
	414	+ end
	415	+
	416	+ # A drawn shape can run to a thousand points; at the weight these are
	417	+ # rendered, far fewer is indistinguishable and keeps the payload sane when a
	418	+ # feed has hundreds of routes.
	419	+ @max_line_points 200
	420	+
	421	+ @doc """
	422	+ One polyline per route: `[%{id:, color:, points: [[lat, lng], ...]}]`.
	423	+
	424	+ Uses shapes.txt where the feed ships it, which is the real drawn alignment.
	425	+ Routes with no shape fall back to the ordered stops of a representative trip
	426	+ -- straight hops between stops rather than following the road, but better
	427	+ than nothing on feeds that omit shapes.
	428	+ """
	429	+ def list_route_lines(source_ids) when is_list(source_ids) do
	430	+ source_ids
	431	+ \|> Enum.uniq()
	432	+ \|> Enum.flat_map(fn source_id ->
	433	+ # route_id is only unique within a feed, so the map would otherwise get
	434	+ # two lines claiming the same DOM id.
	435	+ source_id
	436	+ \|> list_route_lines()
	437	+ \|> Enum.map(fn line -> %{line \| id: "#{source_id}-#{line.id}"} end)
	438	+ end)
	439	+ end
	440	+
	441	+ def list_route_lines(source_id) do
	442	+ shaped = shaped_route_lines(source_id)
	443	+ covered = MapSet.new(shaped, & &1.id)
	444	+
	445	+ uncovered =
	446	+ from(r in Route, where: r.source_id == ^source_id, select: r.route_id)
	447	+ \|> Repo.all()
	448	+ \|> Enum.reject(&MapSet.member?(covered, &1))
	449	+
	450	+ # The stop-sequence fallback reads stop_times, which runs to millions of
	451	+ # rows; it is only worth paying for the routes shapes did not cover, and
	452	+ # on a feed with complete shapes that is none of them.
	453	+ shaped ++ stop_route_lines(source_id, uncovered)
	454	+ end
	455	+
	456	+ defp shaped_route_lines(source_id) do
	457	+ Repo.query!(
	458	+ """
	459	+ WITH rep AS (
	460	+ SELECT DISTINCT ON (t.route_id) t.route_id, t.shape_id
	461	+ FROM gtfs_trips t
	462	+ WHERE t.source_id = $1 AND t.shape_id IS NOT NULL AND t.shape_id <> ''
	463	+ ORDER BY t.route_id, t.shape_id
	464	+ )
	465	+ SELECT r.route_id, r.route_color, sh.shape_pt_lat, sh.shape_pt_lon
	466	+ FROM rep
	467	+ JOIN gtfs_routes r ON r.source_id = $1 AND r.route_id = rep.route_id
	468	+ JOIN gtfs_shapes sh ON sh.source_id = $1 AND sh.shape_id = rep.shape_id
	469	+ WHERE sh.shape_pt_lat IS NOT NULL AND sh.shape_pt_lon IS NOT NULL
	470	+ ORDER BY r.route_id, sh.shape_pt_sequence
	471	+ """,
	472	+ [source_id]
	473	+ )
	474	+ \|> rows_to_lines()
	475	+ end
	476	+
	477	+ defp stop_route_lines(_source_id, []), do: []
	478	+
	479	+ defp stop_route_lines(source_id, route_ids) do
	480	+ Repo.query!(
	481	+ """
	482	+ WITH rep AS (
	483	+ SELECT DISTINCT ON (t.route_id) t.route_id, t.trip_id
	484	+ FROM gtfs_trips t
	485	+ WHERE t.source_id = $1 AND t.route_id = ANY($2)
	486	+ ORDER BY t.route_id, t.trip_id
	487	+ )
	488	+ SELECT r.route_id, r.route_color, s.stop_lat, s.stop_lon
	489	+ FROM rep
	490	+ JOIN gtfs_routes r ON r.source_id = $1 AND r.route_id = rep.route_id
	491	+ JOIN gtfs_stop_times st ON st.source_id = $1 AND st.trip_id = rep.trip_id
	492	+ JOIN gtfs_stops s ON s.source_id = $1 AND s.stop_id = st.stop_id
	493	+ WHERE s.stop_lat IS NOT NULL AND s.stop_lon IS NOT NULL
	494	+ ORDER BY r.route_id, st.stop_sequence
	495	+ """,
	496	+ [source_id, route_ids]
	497	+ )
	498	+ \|> rows_to_lines()
	499	+ end
	500	+
	501	+ defp rows_to_lines(%{rows: rows}) do
	502	+ rows
	503	+ \|> Enum.group_by(fn [route_id, _color, _lat, _lon] -> route_id end)
	504	+ \|> Enum.map(fn {route_id, grouped} ->
	505	+ [[_, color, _, _] \| _] = grouped
	506	+
	507	+ %{
	508	+ id: route_id,
	509	+ color: route_color(color),
	510	+ points: grouped \|> Enum.map(fn [_, _, lat, lon] -> [lat, lon] end) \|> decimate()
	511	+ }
	512	+ end)
	513	+ # A single point is not a line, and a route with no usable geometry is noise.
	514	+ \|> Enum.filter(&(length(&1.points) > 1))
	515	+ end
	516	+
	517	+ # Every nth point, with the last one always kept so the line does not stop
	518	+ # short of where the route actually ends.
	519	+ defp decimate(points) do
	520	+ case length(points) do
	521	+ n when n <= @max_line_points ->
	522	+ points
	523	+
	524	+ n ->
	525	+ step = div(n, @max_line_points) + 1
	526	+
	527	+ kept =
	528	+ points
	529	+ \|> Enum.with_index()
	530	+ \|> Enum.filter(fn {_point, index} -> rem(index, step) == 0 end)
	531	+ \|> Enum.map(&elem(&1, 0))
	532	+
	533	+ last = List.last(points)
	534	+ if List.last(kept) == last, do: kept, else: kept ++ [last]
	535	+ end
	536	+ end
	537	+
	538	+ # GTFS stores route_color bare ("FFC72C"); CSS needs the hash.
	539	+ defp route_color(nil), do: nil
	540	+ defp route_color(""), do: nil
	541	+ defp route_color("#" <> _ = color), do: color
	542	+ defp route_color(color), do: "#" <> color
	543	+
380	544	def count_routes(source_id) do
381	545	Repo.one(from p in Route, where: p.source_id == ^source_id, select: count(p.id))
382	546	end

@@ -458,6 +622,16 @@ defmodule RoomSanctum.Storage do

458	622	Repo.delete(route)
459	623	end
460	624
	625	+ def truncate_shape(source_id) do
	626	+ from(p in Shape, where: p.source_id == ^source_id)
	627	+ \|> Repo.delete_all()
	628	+ end
	629	+
	630	+ def count_shapes(source_id) do
	631	+ from(p in Shape, where: p.source_id == ^source_id, select: count(p.id))
	632	+ \|> Repo.one()
	633	+ end
	634	+
461	635	def truncate_route(source_id) do
462	636	from(p in Route, where: p.source_id == ^source_id)
463	637	\|> Repo.delete_all()

added apps/room_sanctum/lib/room_sanctum/storage/gtfs/shape.ex

+36 −0

@@ -0,0 +1,36 @@

	1	+defmodule RoomSanctum.Storage.GTFS.Shape do
	2	+ use Ecto.Schema
	3	+ import Ecto.Changeset
	4	+
	5	+ @moduledoc """
	6	+ A point on a route's drawn path, from shapes.txt.
	7	+
	8	+ Trips reference these by shape_id, which is what makes a route line follow
	9	+ the road rather than hopping stop to stop.
	10	+ """
	11	+
	12	+ schema "gtfs_shapes" do
	13	+ belongs_to :source, RoomSanctum.Configuration.Source
	14	+ field :shape_id, :string
	15	+ field :shape_pt_lat, :float
	16	+ field :shape_pt_lon, :float
	17	+ field :shape_pt_sequence, :integer
	18	+ field :shape_dist_traveled, :float
	19	+
	20	+ timestamps()
	21	+ end
	22	+
	23	+ @doc false
	24	+ def changeset(shape, attrs) do
	25	+ shape
	26	+ \|> cast(attrs, [
	27	+ :shape_id,
	28	+ :shape_pt_lat,
	29	+ :shape_pt_lon,
	30	+ :shape_pt_sequence,
	31	+ :shape_dist_traveled,
	32	+ :source_id
	33	+ ])
	34	+ \|> validate_required([:shape_id, :source_id])
	35	+ end
	36	+end

added apps/room_sanctum/priv/repo/migrations/20260810180000_create_gtfs_shapes.exs

+22 −0

@@ -0,0 +1,22 @@

	1	+defmodule RoomSanctum.Repo.Migrations.CreateGtfsShapes do
	2	+ use Ecto.Migration
	3	+
	4	+ def change do
	5	+ create table(:gtfs_shapes) do
	6	+ add :source_id, references(:cfg_sources, on_delete: :delete_all), null: false
	7	+
	8	+ add :shape_id, :string, null: false
	9	+ add :shape_pt_lat, :float
	10	+ add :shape_pt_lon, :float
	11	+ add :shape_pt_sequence, :integer
	12	+ # Optional in the spec, and MBTA ships it empty.
	13	+ add :shape_dist_traveled, :float
	14	+
	15	+ timestamps()
	16	+ end
	17	+
	18	+ # The only read is "give me one shape's points in order", which this covers
	19	+ # as a prefix scan.
	20	+ create index(:gtfs_shapes, [:source_id, :shape_id, :shape_pt_sequence])
	21	+ end
	22	+end

added apps/room_sanctum/priv/repo/migrations/20260810190000_index_stop_times_by_stop.exs

+10 −0

@@ -0,0 +1,10 @@

	1	+defmodule RoomSanctum.Repo.Migrations.IndexStopTimesByStop do
	2	+ use Ecto.Migration
	3	+
	4	+ def change do
	5	+ # "which routes call at this stop" reads stop_times by stop_id, but the
	6	+ # only index there leads with trip_id, so the planner was scanning every
	7	+ # trip in the feed and probing per row.
	8	+ create index(:gtfs_stop_times, [:source_id, :stop_id])
	9	+ end
	10	+end

added apps/room_sanctum/test/room_sanctum/gtfs_route_lines_test.exs

+252 −0

@@ -0,0 +1,252 @@

	1	+defmodule RoomSanctum.GtfsRouteLinesTest do
	2	+ @moduledoc """
	3	+ Route geometry for the map's route-line layer, and the linked_datasets.txt
	4	+ parsing that wires up GTFS-RT automatically.
	5	+ """
	6	+ use RoomSanctum.DataCase
	7	+
	8	+ alias RoomSanctum.{Accounts, Configuration, Repo, Storage}
	9	+ alias RoomSanctum.Storage.GTFS.{Route, Shape, Stop, StopTime, Trip}
	10	+
	11	+ setup do
	12	+ {:ok, user} =
	13	+ Accounts.register_user(%{
	14	+ email: "lines#{System.unique_integer([:positive])}@example.com",
	15	+ password: "hello world!hello world!"
	16	+ })
	17	+
	18	+ {:ok, source} =
	19	+ Configuration.create_source(%{
	20	+ name: "Feed",
	21	+ notes: "",
	22	+ type: :gtfs,
	23	+ enabled: true,
	24	+ user_id: user.id,
	25	+ config: %{"__type__" => "gtfs", "url" => "https://example.test/g.zip", "tz" => "UTC"}
	26	+ })
	27	+
	28	+ %{source: source, sid: source.id}
	29	+ end
	30	+
	31	+ defp route(sid, id, opts \\ []) do
	32	+ Repo.insert!(%Route{
	33	+ source_id: sid,
	34	+ route_id: id,
	35	+ route_type: Keyword.get(opts, :type, "3"),
	36	+ route_color: Keyword.get(opts, :color)
	37	+ })
	38	+ end
	39	+
	40	+ defp trip(sid, route_id, trip_id, shape_id \\ nil) do
	41	+ Repo.insert!(%Trip{
	42	+ source_id: sid,
	43	+ route_id: route_id,
	44	+ trip_id: trip_id,
	45	+ shape_id: shape_id,
	46	+ service_id: "svc"
	47	+ })
	48	+ end
	49	+
	50	+ defp shape_points(sid, shape_id, count) do
	51	+ for i <- 1..count do
	52	+ Repo.insert!(%Shape{
	53	+ source_id: sid,
	54	+ shape_id: shape_id,
	55	+ shape_pt_lat: 37.0 + i / 10_000,
	56	+ shape_pt_lon: -122.0,
	57	+ shape_pt_sequence: i
	58	+ })
	59	+ end
	60	+ end
	61	+
	62	+ defp stops_for(sid, trip_id, count) do
	63	+ for i <- 1..count do
	64	+ stop_id = "#{trip_id}-s#{i}"
	65	+ Repo.insert!(%Stop{source_id: sid, stop_id: stop_id, stop_name: stop_id,
	66	+ stop_lat: 38.0 + i / 1000, stop_lon: -122.0})
	67	+ Repo.insert!(%StopTime{source_id: sid, trip_id: trip_id, stop_id: stop_id,
	68	+ stop_sequence: i})
	69	+ end
	70	+ end
	71	+
	72	+ defp line_for(lines, id), do: Enum.find(lines, &(&1.id == id))
	73	+
	74	+ describe "route geometry" do
	75	+ test "a route with a shape uses the drawn alignment", %{sid: sid} do
	76	+ route(sid, "R1")
	77	+ trip(sid, "R1", "T1", "SH1")
	78	+ shape_points(sid, "SH1", 12)
	79	+ # stops exist too, and must lose to the shape
	80	+ stops_for(sid, "T1", 3)
	81	+
	82	+ line = Storage.list_route_lines(sid) \|> line_for("R1")
	83	+
	84	+ assert length(line.points) == 12
	85	+ assert [lat, lon] = hd(line.points)
	86	+ # shapes sit at 37.x, the stop fallback at 38.x
	87	+ assert_in_delta lat, 37.0, 0.01
	88	+ assert lon == -122.0
	89	+ end
	90	+
	91	+ test "a route with no shape falls back to its stop sequence", %{sid: sid} do
	92	+ route(sid, "R2")
	93	+ trip(sid, "R2", "T2", nil)
	94	+ stops_for(sid, "T2", 4)
	95	+
	96	+ line = Storage.list_route_lines(sid) \|> line_for("R2")
	97	+
	98	+ assert length(line.points) == 4
	99	+ assert [lat, _lon] = hd(line.points)
	100	+ assert_in_delta lat, 38.0, 0.01
	101	+ end
	102	+
	103	+ test "shaped and unshaped routes coexist in one feed", %{sid: sid} do
	104	+ route(sid, "SHAPED")
	105	+ trip(sid, "SHAPED", "TA", "SHA")
	106	+ shape_points(sid, "SHA", 5)
	107	+
	108	+ route(sid, "PLAIN")
	109	+ trip(sid, "PLAIN", "TB", nil)
	110	+ stops_for(sid, "TB", 3)
	111	+
	112	+ lines = Storage.list_route_lines(sid)
	113	+
	114	+ assert length(lines) == 2
	115	+ assert length(line_for(lines, "SHAPED").points) == 5
	116	+ assert length(line_for(lines, "PLAIN").points) == 3
	117	+ end
	118	+
	119	+ test "an empty shape_id is treated as no shape", %{sid: sid} do
	120	+ route(sid, "R3")
	121	+ trip(sid, "R3", "T3", "")
	122	+ stops_for(sid, "T3", 3)
	123	+
	124	+ assert length(Storage.list_route_lines(sid) \|> line_for("R3") \|> Map.fetch!(:points)) == 3
	125	+ end
	126	+
	127	+ test "a long shape is decimated but still ends where the route ends", %{sid: sid} do
	128	+ route(sid, "LONG")
	129	+ trip(sid, "LONG", "TL", "SHL")
	130	+ shape_points(sid, "SHL", 1000)
	131	+
	132	+ line = Storage.list_route_lines(sid) \|> line_for("LONG")
	133	+
	134	+ assert length(line.points) <= 200, "got #{length(line.points)} points"
	135	+ assert length(line.points) > 50, "decimated too far: #{length(line.points)}"
	136	+ # the 1000th point, i.e. the true end of the line
	137	+ assert List.last(line.points) == [37.0 + 1000 / 10_000, -122.0]
	138	+ assert hd(line.points) == [37.0 + 1 / 10_000, -122.0]
	139	+ end
	140	+
	141	+ test "a one-point shape is not a line", %{sid: sid} do
	142	+ route(sid, "DOT")
	143	+ trip(sid, "DOT", "TD", "SHD")
	144	+ shape_points(sid, "SHD", 1)
	145	+
	146	+ assert Storage.list_route_lines(sid) \|> line_for("DOT") == nil
	147	+ end
	148	+
	149	+ test "route_color gains the hash CSS needs, and absence is tolerated", %{sid: sid} do
	150	+ route(sid, "BARE", color: "FFC72C")
	151	+ trip(sid, "BARE", "T4", "SH4")
	152	+ shape_points(sid, "SH4", 3)
	153	+
	154	+ route(sid, "HASHED", color: "#0a5")
	155	+ trip(sid, "HASHED", "T5", "SH5")
	156	+ shape_points(sid, "SH5", 3)
	157	+
	158	+ route(sid, "NONE", color: nil)
	159	+ trip(sid, "NONE", "T6", "SH6")
	160	+ shape_points(sid, "SH6", 3)
	161	+
	162	+ lines = Storage.list_route_lines(sid)
	163	+
	164	+ assert line_for(lines, "BARE").color == "#FFC72C"
	165	+ assert line_for(lines, "HASHED").color == "#0a5"
	166	+ assert line_for(lines, "NONE").color == nil
	167	+ end
	168	+
	169	+ test "a feed with no geometry at all yields nothing rather than raising", %{sid: sid} do
	170	+ route(sid, "EMPTY")
	171	+ assert Storage.list_route_lines(sid) == []
	172	+ end
	173	+
	174	+ test "ids stay unique when several feeds are drawn on one map", %{sid: sid, source: source} do
	175	+ route(sid, "1")
	176	+ trip(sid, "1", "T1", "SH1")
	177	+ shape_points(sid, "SH1", 3)
	178	+
	179	+ {:ok, other} =
	180	+ Configuration.create_source(%{
	181	+ name: "Other feed",
	182	+ notes: "",
	183	+ type: :gtfs,
	184	+ enabled: true,
	185	+ user_id: source.user_id,
	186	+ config: %{"__type__" => "gtfs", "url" => "https://example.test/o.zip", "tz" => "UTC"}
	187	+ })
	188	+
	189	+ # same route_id in a different feed -- the ids must not collide
	190	+ route(other.id, "1")
	191	+ trip(other.id, "1", "T1", "SH1")
	192	+ shape_points(other.id, "SH1", 3)
	193	+
	194	+ ids = Storage.list_route_lines([sid, other.id]) \|> Enum.map(& &1.id) \|> Enum.sort()
	195	+
	196	+ assert ids == Enum.sort(["#{other.id}-1", "#{sid}-1"])
	197	+ end
	198	+ end
	199	+
	200	+ describe "routes serviced by a stop" do
	201	+ setup %{sid: sid} do
	202	+ for stop_id <- ["shared", "only_a", "only_b"] do
	203	+ Repo.insert!(%Stop{source_id: sid, stop_id: stop_id, stop_name: stop_id,
	204	+ stop_lat: 37.0, stop_lon: -122.0})
	205	+ end
	206	+
	207	+ # A calls at shared + only_a, B at shared + only_b, C at only_b alone
	208	+ for {rid, stops} <- [{"A", ["shared", "only_a"]}, {"B", ["shared", "only_b"]}, {"C", ["only_b"]}] do
	209	+ route(sid, rid)
	210	+ trip(sid, rid, "T#{rid}", nil)
	211	+
	212	+ for {stop_id, seq} <- Enum.with_index(stops) do
	213	+ Repo.insert!(%StopTime{source_id: sid, trip_id: "T#{rid}", stop_id: stop_id,
	214	+ stop_sequence: seq})
	215	+ end
	216	+ end
	217	+
	218	+ :ok
	219	+ end
	220	+
	221	+ test "returns every route calling there, and only those", %{sid: sid} do
	222	+ assert Storage.routes_serving_stop(sid, "shared") \|> Enum.sort() == ["A", "B"]
	223	+ assert Storage.routes_serving_stop(sid, "only_a") \|> Enum.sort() == ["A"]
	224	+ assert Storage.routes_serving_stop(sid, "only_b") \|> Enum.sort() == ["B", "C"]
	225	+ end
	226	+
	227	+ test "a route calling twice is listed once", %{sid: sid} do
	228	+ Repo.insert!(%StopTime{source_id: sid, trip_id: "TA", stop_id: "shared", stop_sequence: 99})
	229	+
	230	+ assert Storage.routes_serving_stop(sid, "shared") \|> Enum.sort() == ["A", "B"]
	231	+ end
	232	+
	233	+ test "an unknown stop serves nothing", %{sid: sid} do
	234	+ assert Storage.routes_serving_stop(sid, "nope") == []
	235	+ end
	236	+
	237	+ test "does not leak across feeds", %{sid: sid, source: source} do
	238	+ {:ok, other} =
	239	+ Configuration.create_source(%{
	240	+ name: "Elsewhere",
	241	+ notes: "",
	242	+ type: :gtfs,
	243	+ enabled: true,
	244	+ user_id: source.user_id,
	245	+ config: %{"__type__" => "gtfs", "url" => "https://example.test/e.zip", "tz" => "UTC"}
	246	+ })
	247	+
	248	+ assert Storage.routes_serving_stop(other.id, "shared") == []
	249	+ assert sid != other.id
	250	+ end
	251	+ end
	252	+end

Parents: 0a0ddd6