move route_type cache to separate module instead of relying on trip_i…

…d to check route type

lib/concentrate/group_filter/cancelled_trip.ex

@@ -4,16 +4,16 @@ defmodule Concentrate.GroupFilter.CancelledTrip do
   """
   @behaviour Concentrate.GroupFilter
   alias Concentrate.Filter.Alert.CancelledTrips
-  alias Concentrate.GTFS.Trips
+  alias Concentrate.GTFS.Routes
   alias Concentrate.{StopTimeUpdate, TripDescriptor}
 
   @impl Concentrate.GroupFilter
-  def filter(trip_group, module \\ CancelledTrips, trips_module \\ Trips)
+  def filter(trip_group, module \\ CancelledTrips, routes_module \\ Routes)
 
   def filter(
         {%TripDescriptor{} = td, _vps, [stu | _] = stop_time_updates} = group,
         module,
-        trips_module
+        routes_module
       ) do
     trip_id = TripDescriptor.trip_id(td)
     route_id = TripDescriptor.route_id(td)
@@ -23,7 +23,7 @@ defmodule Concentrate.GroupFilter.CancelledTrip do
       TripDescriptor.schedule_relationship(td) == :CANCELED ->
         cancel_group(group)
 
-      bus_block_waiver?(stop_time_updates, trips_module.route_type(trip_id)) ->
+      bus_block_waiver?(stop_time_updates, routes_module.route_type(trip_id)) ->
         cancel_group(group)
 
       is_nil(time) ->

lib/concentrate/gtfs/routes.ex

@@ -0,0 +1,48 @@
+defmodule Concentrate.GTFS.Routes do
+  @moduledoc """
+  Server which maintains a list of route_id -> route_type mappings.
+  """
+  use GenStage
+  require Logger
+  import :binary, only: [copy: 1]
+  @table __MODULE__
+
+  def start_link(opts) do
+    GenStage.start_link(__MODULE__, opts, name: __MODULE__)
+  end
+
+  def route_type(route_id) do
+    hd(:ets.lookup_element(@table, route_id, 2))
+  rescue
+    ArgumentError -> nil
+  end
+
+  def init(opts) do
+    @table = :ets.new(@table, [:named_table, :public, :duplicate_bag])
+    {:consumer, %{}, opts}
+  end
+
+  def handle_events(events, _from, state) do
+    inserts =
+      for event <- events,
+          {"routes.txt", route_body} <- event,
+          lines = String.split(route_body, "\n"),
+          {:ok, row} <- CSV.decode(lines, headers: true) do
+        {copy(row["route_id"]), String.to_integer(row["route_type"])}
+      end
+
+    _ =
+      if inserts == [] do
+        :ok
+      else
+        true = :ets.delete_all_objects(@table)
+        :ets.insert(@table, inserts)
+
+        Logger.info(fn ->
+          "#{__MODULE__}: updated with #{length(inserts)} records"
+        end)
+      end
+
+    {:noreply, [], state, :hibernate}
+  end
+end

lib/concentrate/gtfs/trips.ex

@@ -1,6 +1,6 @@
 defmodule Concentrate.GTFS.Trips do
   @moduledoc """
-  Server which maintains a list of trip -> {route_id, direction_id, route_type} mappings.
+  Server which maintains a list of trip -> {route_id, direction_id} mappings.
   """
   use GenStage
   require Logger
@@ -23,34 +23,18 @@ defmodule Concentrate.GTFS.Trips do
     ArgumentError -> nil
   end
 
-  def route_type(trip_id) do
-    hd(:ets.lookup_element(@table, trip_id, 4))
-  rescue
-    ArgumentError -> nil
-  end
-
   def init(opts) do
     @table = :ets.new(@table, [:named_table, :public, :duplicate_bag])
     {:consumer, %{}, opts}
   end
 
   def handle_events(events, _from, state) do
-    route_id_type_map =
-      for event <- events,
-          {"routes.txt", trip_body} <- event,
-          lines = String.split(trip_body, "\n"),
-          {:ok, row} <- CSV.decode(lines, headers: true) do
-        {copy(row["route_id"]), String.to_integer(row["route_type"])}
-      end
-      |> Map.new()
-
     inserts =
       for event <- events,
           {"trips.txt", trip_body} <- event,
           lines = String.split(trip_body, "\n"),
           {:ok, row} <- CSV.decode(lines, headers: true) do
-        {copy(row["trip_id"]), copy(row["route_id"]), String.to_integer(row["direction_id"]),
-         route_id_type_map[row["route_id"]]}
+        {copy(row["trip_id"]), copy(row["route_id"]), String.to_integer(row["direction_id"])}
       end
 
     _ =

test/concentrate/group_filter/cancelled_trip_test.exs

@@ -5,7 +5,7 @@ defmodule Concentrate.GroupFilter.CancelledTripTest do
   alias Concentrate.{StopTimeUpdate, TripDescriptor}
 
   @module Concentrate.Filter.FakeCancelledTrips
-  @fake_trips_module Concentrate.GTFS.FakeTrips
+  @fake_routes_module Concentrate.GTFS.FakeRoutes
 
   describe "filter/2" do
     test "cancels the group if the trip is cancelled by an alert" do
@@ -22,7 +22,7 @@ defmodule Concentrate.GroupFilter.CancelledTripTest do
         )
 
       group = {td, [], [stu]}
-      {new_td, [], [new_stu]} = filter(group, @module, @fake_trips_module)
+      {new_td, [], [new_stu]} = filter(group, @module, @fake_routes_module)
       assert TripDescriptor.schedule_relationship(new_td) == :CANCELED
       assert StopTimeUpdate.schedule_relationship(new_stu) == :SKIPPED
     end
@@ -42,7 +42,7 @@ defmodule Concentrate.GroupFilter.CancelledTripTest do
         )
 
       group = {td, [], [stu]}
-      {new_td, [], [new_stu]} = filter(group, @module, @fake_trips_module)
+      {new_td, [], [new_stu]} = filter(group, @module, @fake_routes_module)
       assert TripDescriptor.schedule_relationship(new_td) == :CANCELED
       assert StopTimeUpdate.schedule_relationship(new_stu) == :SKIPPED
     end
@@ -62,7 +62,7 @@ defmodule Concentrate.GroupFilter.CancelledTripTest do
         )
 
       group = {td, [], [stu]}
-      {_td, [], [stu]} = filter(group, @module, @fake_trips_module)
+      {_td, [], [stu]} = filter(group, @module, @fake_routes_module)
       assert StopTimeUpdate.schedule_relationship(stu) == :SKIPPED
     end
 
@@ -82,7 +82,7 @@ defmodule Concentrate.GroupFilter.CancelledTripTest do
         )
 
       group = {td, [], [stu, stu]}
-      {td_actual, [], [stu_actual1, stu_actual2]} = filter(group, @module, @fake_trips_module)
+      {td_actual, [], [stu_actual1, stu_actual2]} = filter(group, @module, @fake_routes_module)
       assert TripDescriptor.schedule_relationship(td_actual) == :CANCELED
       assert StopTimeUpdate.schedule_relationship(stu_actual1) == :SKIPPED
       assert StopTimeUpdate.schedule_relationship(stu_actual2) == :SKIPPED
@@ -104,7 +104,7 @@ defmodule Concentrate.GroupFilter.CancelledTripTest do
         )
 
       group = {td, [], [stu, stu]}
-      {td_actual, [], [stu_actual1, stu_actual2]} = filter(group, @module, @fake_trips_module)
+      {td_actual, [], [stu_actual1, stu_actual2]} = filter(group, @module, @fake_routes_module)
       assert TripDescriptor.schedule_relationship(td_actual) == :SCHEDULED
       assert StopTimeUpdate.schedule_relationship(stu_actual1) == :SKIPPED
       assert StopTimeUpdate.schedule_relationship(stu_actual2) == :SKIPPED
@@ -132,7 +132,7 @@ defmodule Concentrate.GroupFilter.CancelledTripTest do
         )
 
       group = {td, [], [stu1, stu2]}
-      {td_actual, [], [stu_actual1, stu_actual2]} = filter(group, @module, @fake_trips_module)
+      {td_actual, [], [stu_actual1, stu_actual2]} = filter(group, @module, @fake_routes_module)
       assert TripDescriptor.schedule_relationship(td_actual) == :SCHEDULED
       assert StopTimeUpdate.schedule_relationship(stu_actual1) == :SKIPPED
       assert StopTimeUpdate.schedule_relationship(stu_actual2) == :SCHEDULED
@@ -152,7 +152,7 @@ defmodule Concentrate.GroupFilter.CancelledTripTest do
         )
 
       group = {td, [], [stu]}
-      assert filter(group, @module, @fake_trips_module) == group
+      assert filter(group, @module, @fake_routes_module) == group
     end
 
     test "does not cancel the group if the route was cancelled at a different time" do
@@ -170,7 +170,7 @@ defmodule Concentrate.GroupFilter.CancelledTripTest do
         )
 
       group = {td, [], [stu]}
-      assert filter(group, @module, @fake_trips_module) == group
+      assert filter(group, @module, @fake_routes_module) == group
     end
 
     test "other values are returned as-is" do

test/concentrate/gtfs/routes_test.exs

@@ -0,0 +1,37 @@
+defmodule Concentrate.GTFS.RoutesTest do
+  @moduledoc false
+  use ExUnit.Case
+  import Concentrate.GTFS.Routes
+
+  @body """
+  route_id,route_type
+  route_1,0
+  route_2,1
+  route_3,2
+  """
+
+  defp supervised(_) do
+    start_supervised(Concentrate.GTFS.Routes)
+    event = [{"routes.txt", @body}]
+    # relies on being able to update the table from a different process
+    handle_events([event], :ignored, :ignored)
+    :ok
+  end
+
+  describe "route_type/1" do
+    setup :supervised
+
+    test "returns the route_type for the given route_id" do
+      assert route_type("route_1") == 0
+      assert route_type("route_2") == 1
+      assert route_type("route_3") == 2
+      assert route_type("unknown") == nil
+    end
+  end
+
+  describe "missing ETS table" do
+    test "route_type/1 returns nil" do
+      assert route_type("route_1") == nil
+    end
+  end
+end

test/concentrate/gtfs/trips_test.exs

@@ -8,14 +8,9 @@ defmodule Concentrate.GTFS.TripsTest do
   trip,route,5
   """
 
-  @routes_body """
-  route_id,route_type
-  route,3
-  """
-
   defp supervised(_) do
     start_supervised(Concentrate.GTFS.Trips)
-    event = [{"trips.txt", @body}, {"routes.txt", @routes_body}]
+    event = [{"trips.txt", @body}]
     # relies on being able to update the table from a different process
     handle_events([event], :ignored, :ignored)
     :ok
@@ -39,15 +34,6 @@ defmodule Concentrate.GTFS.TripsTest do
     end
   end
 
-  describe "route_type/1" do
-    setup :supervised
-
-    test "returns the direction_id for the given trip" do
-      assert route_type("trip") == 3
-      assert route_type("unknown") == nil
-    end
-  end
-
   describe "missing ETS table" do
     test "route_id/1 returns nil" do
       assert route_id("trip") == nil

test/support/filter/fakes.ex

@@ -5,7 +5,10 @@ defmodule Concentrate.GTFS.FakeTrips do
 
   def direction_id("trip"), do: 1
   def direction_id(_), do: nil
+end
 
+defmodule Concentrate.GTFS.FakeRoutes do
+  @moduledoc "Fake implementation of GTFS.Routes"
   def route_type("trip"), do: 3
   def route_type(_), do: nil
 end