fix: require stop_sequence for StopTimeUpdates

Previously we merged StopTimeUpdates using a "key" of trip ID + stop
ID + stop sequence. Since stop ID and sequence are both optional (only
one needs to be present), this created the potential for duplicates:
two updates that referred to the same stop would not be merged if one
of them used a sequence and the other used a stop ID.

Given:

1. it's not universally possible to merge StopTimeUpdates without a
   stop sequence (it's required if a trip contains a loop)
2. merging updates correctly using "maybe stop ID or sequence or both"
   would require significant changes to the merging system
3. grouping and the `RemoveUnneededTimes` filter assume the presence of
   stop sequences
4. all our current input feeds either include stop sequences already or
   are planned to

...we can simplify things by explicitly requiring a stop sequence, and
only merging based on trip ID + stop sequence.

config/config.exs

@@ -23,6 +23,7 @@ config :concentrate,
     url: "https://cdn.mbta.com/MBTA_GTFS.zip"
   ],
   filters: [
+    Concentrate.Filter.NullStopSequence,
     Concentrate.Filter.VehicleWithNoTrip,
     Concentrate.Filter.RoundSpeedAndBearing,
     Concentrate.Filter.IncludeRouteDirection,

lib/concentrate/filter/null_stop_sequence.ex

@@ -0,0 +1,19 @@
+defmodule Concentrate.Filter.NullStopSequence do
+  @moduledoc """
+  Filters out StopTimeUpdates with a null `stop_sequence`, since they would not have been merged
+  correctly (and wouldn't work with downstream modules that assume a stop sequence is present).
+  """
+  @behaviour Concentrate.Filter
+  alias Concentrate.StopTimeUpdate
+  require Logger
+
+  @impl Concentrate.Filter
+  def filter(%StopTimeUpdate{} = stu) do
+    case StopTimeUpdate.stop_sequence(stu) do
+      nil -> :skip
+      _ -> {:cont, stu}
+    end
+  end
+
+  def filter(other), do: {:cont, other}
+end

lib/concentrate/stop_time_update.ex

@@ -3,7 +3,6 @@ defmodule Concentrate.StopTimeUpdate do
   Structure for representing an update to a StopTime (e.g. a predicted arrival or departure)
   """
   import Concentrate.StructHelpers
-  alias Concentrate.Filter.GTFS.Stops
 
   defstruct_accessors([
     :trip_id,
@@ -36,14 +35,7 @@ defmodule Concentrate.StopTimeUpdate do
   end
 
   defimpl Concentrate.Mergeable do
-    def key(%{trip_id: trip_id, stop_id: stop_id, stop_sequence: stop_sequence}) do
-      parent_station_id =
-        if stop_id do
-          Stops.parent_station_id(stop_id)
-        end
-
-      {trip_id, parent_station_id, stop_sequence}
-    end
+    def key(%{trip_id: trip_id, stop_sequence: stop_sequence}), do: {trip_id, stop_sequence}
 
     def merge(first, second) do
       %{

test/concentrate/filter/null_stop_sequence_test.exs

@@ -0,0 +1,22 @@
+defmodule Concentrate.Filter.NullStopSequenceTest do
+  @moduledoc false
+  use ExUnit.Case, async: true
+  import Concentrate.Filter.NullStopSequence
+  alias Concentrate.StopTimeUpdate
+
+  describe "filter/1" do
+    test "skips a stop time update with a null stop_sequence" do
+      stu = StopTimeUpdate.new(stop_sequence: nil)
+      assert :skip = filter(stu)
+    end
+
+    test "passes through a stop time update with a stop_sequence" do
+      stu = StopTimeUpdate.new(stop_sequence: 1)
+      assert {:cont, ^stu} = filter(stu)
+    end
+
+    test "passes through other values" do
+      assert {:cont, :value} = filter(:value)
+    end
+  end
+end

test/concentrate/stop_time_update_test.exs

@@ -6,7 +6,6 @@ defmodule Concentrate.StopTimeUpdateTest do
 
   @stu new(
          trip_id: "trip",
-         stop_id: "stop",
          stop_sequence: 1,
          arrival_time: 2,
          departure_time: 3,
@@ -28,22 +27,13 @@ defmodule Concentrate.StopTimeUpdateTest do
   end
 
   describe "Concentrate.Mergeable" do
-    test "key/1 uses the parent station ID" do
-      start_supervised!(Concentrate.Filter.GTFS.Stops)
-      Concentrate.Filter.GTFS.Stops._insert_mapping("child_id", "parent_id")
-
-      assert Mergeable.key(new(stop_id: "child_id")) == {nil, "parent_id", nil}
-      assert Mergeable.key(new(stop_id: "other")) == {nil, "other", nil}
-      assert Mergeable.key(new(stop_id: nil)) == {nil, nil, nil}
-    end
-
-    test "merge/2 takes non-nil values, earliest arrival, latest departure" do
+    test "takes non-nil values, earliest arrival, latest departure" do
       first = @stu
 
       second =
         new(
           trip_id: "trip",
-          stop_id: "stop-01",
+          stop_id: "stop",
           stop_sequence: 1,
           arrival_time: 1,
           departure_time: 4,
@@ -55,7 +45,7 @@ defmodule Concentrate.StopTimeUpdateTest do
       expected =
         new(
           trip_id: "trip",
-          stop_id: "stop-01",
+          stop_id: "stop",
           stop_sequence: 1,
           arrival_time: 1,
           departure_time: 4,
@@ -69,5 +59,13 @@ defmodule Concentrate.StopTimeUpdateTest do
       assert Mergeable.merge(first, second) == expected
       assert Mergeable.merge(second, first) == expected
     end
+
+    test "picks the 'greater' of two stop IDs" do
+      first = new(stop_id: "stop")
+      second = new(stop_id: "stop-01")
+
+      assert %{stop_id: "stop-01"} = Mergeable.merge(first, second)
+      assert %{stop_id: "stop-01"} = Mergeable.merge(second, first)
+    end
   end
 end