feat: include passthrough_time from rtr in enhanced trip update feed (#…

…361)

* parse passthrough_time from rtr feed

* do not remove stus if passthrough_time is not nil

* remove stu from normal feed if passthrough_time is present

* clear out arrival and departure times from enhanced feed if passthrough_time is present

lib/concentrate/encoder/trip_updates.ex

@@ -27,7 +27,8 @@ defmodule Concentrate.Encoder.TripUpdates do
 
     relationship = schedule_relationship(StopTimeUpdate.schedule_relationship(update))
 
-    if is_map(arrival) or is_map(departure) or relationship != nil do
+    if (is_map(arrival) or is_map(departure) or relationship != nil) and
+         StopTimeUpdate.passthrough_time(update) == nil do
       drop_nil_values(%{
         stop_id: StopTimeUpdate.stop_id(update),
         stop_sequence: StopTimeUpdate.stop_sequence(update),

lib/concentrate/encoder/trip_updates_enhanced.ex

@@ -29,13 +29,22 @@ defmodule Concentrate.Encoder.TripUpdatesEnhanced do
     drop_nil_values(%{
       stop_id: StopTimeUpdate.stop_id(update),
       stop_sequence: StopTimeUpdate.stop_sequence(update),
-      arrival:
-        stop_time_event(StopTimeUpdate.arrival_time(update), StopTimeUpdate.uncertainty(update)),
-      departure:
-        stop_time_event(StopTimeUpdate.departure_time(update), StopTimeUpdate.uncertainty(update)),
+      arrival: determine_arrival_value(update),
+      departure: determine_departure_value(update),
+      passthrough_time: StopTimeUpdate.passthrough_time(update),
       schedule_relationship: schedule_relationship(StopTimeUpdate.schedule_relationship(update)),
       boarding_status: StopTimeUpdate.status(update),
       platform_id: StopTimeUpdate.platform_id(update)
     })
   end
+
+  defp determine_arrival_value(%{passthrough_time: nil} = update),
+    do: stop_time_event(StopTimeUpdate.arrival_time(update), StopTimeUpdate.uncertainty(update))
+
+  defp determine_arrival_value(_), do: nil
+
+  defp determine_departure_value(%{passthrough_time: nil} = update),
+    do: stop_time_event(StopTimeUpdate.departure_time(update), StopTimeUpdate.uncertainty(update))
+
+  defp determine_departure_value(_), do: nil
 end

lib/concentrate/group_filter/skipped_stop_on_added_trip.ex

@@ -9,7 +9,11 @@ defmodule Concentrate.GroupFilter.SkippedStopOnAddedTrip do
   def filter({%TripDescriptor{} = td, vps, stus}) do
     stus =
       if TripDescriptor.schedule_relationship(td) in ~w(ADDED UNSCHEDULED)a do
-        Enum.reject(stus, &(StopTimeUpdate.schedule_relationship(&1) == :SKIPPED))
+        Enum.reject(
+          stus,
+          &(StopTimeUpdate.schedule_relationship(&1) == :SKIPPED &&
+              StopTimeUpdate.passthrough_time(&1) == nil)
+        )
       else
         stus
       end

lib/concentrate/parser/gtfs_realtime_enhanced.ex

@@ -158,6 +158,7 @@ defmodule Concentrate.Parser.GTFSRealtimeEnhanced do
               schedule_relationship: schedule_relationship(Map.get(stu, "schedule_relationship")),
               arrival_time: arrival_time,
               departure_time: departure_time,
+              passthrough_time: Map.get(stu, "passthrough_time"),
               uncertainty: arrival_uncertainty || departure_uncertainty,
               status: boarding_status,
               platform_id: Map.get(stu, "platform_id")

lib/concentrate/stop_time_update.ex

@@ -9,6 +9,7 @@ defmodule Concentrate.StopTimeUpdate do
     :stop_id,
     :arrival_time,
     :departure_time,
+    :passthrough_time,
     :stop_sequence,
     :status,
     :track,
@@ -56,6 +57,7 @@ defmodule Concentrate.StopTimeUpdate do
         first
         | arrival_time: time_stu.arrival_time,
           departure_time: time_stu.departure_time,
+          passthrough_time: first.passthrough_time || second.passthrough_time,
           status: first.status || second.status,
           track: first.track || second.track,
           schedule_relationship:

test/concentrate/encoder/trip_updates_enhanced_test.exs

@@ -219,5 +219,80 @@ defmodule Concentrate.Encoder.TripUpdatesEnhancedTest do
       trip = get_in(encoded, ["entity", Access.at(0), "trip_update", "trip"])
       refute trip["update_type"]
     end
+
+    test "includes passthrough_time" do
+      parsed = [
+        TripDescriptor.new(
+          trip_id: "trip",
+          route_id: "route",
+          direction_id: 0,
+          schedule_relationship: :ADDED,
+          revenue: true
+        ),
+        StopTimeUpdate.new(
+          trip_id: "trip",
+          stop_id: "stop_1",
+          schedule_relationship: :SKIPPED,
+          departure_time: 100,
+          passthrough_time: 100
+        ),
+        StopTimeUpdate.new(
+          trip_id: "trip",
+          stop_id: "stop_2",
+          schedule_relationship: :SKIPPED,
+          arrival_time: 200,
+          departure_time: 250,
+          passthrough_time: 200
+        ),
+        StopTimeUpdate.new(
+          trip_id: "trip",
+          stop_id: "stop_3",
+          arrival_time: 300,
+          departure_time: 400
+        )
+      ]
+
+      encoded = Jason.decode!(encode_groups(group(parsed)))
+
+      assert %{
+               "entity" => [
+                 %{
+                   "id" => "trip",
+                   "trip_update" => %{
+                     "trip" => %{
+                       "direction_id" => 0,
+                       "revenue" => true,
+                       "route_id" => "route",
+                       "trip_id" => "trip",
+                       "last_trip" => false
+                     },
+                     "stop_time_update" => [
+                       %{
+                         "arrival" => %{"time" => 300},
+                         "departure" => %{"time" => 400},
+                         "stop_id" => "stop_3"
+                       } = stu_3,
+                       %{
+                         "passthrough_time" => 200,
+                         "schedule_relationship" => "SKIPPED",
+                         "stop_id" => "stop_2"
+                       } = stu_2,
+                       %{
+                         "passthrough_time" => 100,
+                         "schedule_relationship" => "SKIPPED",
+                         "stop_id" => "stop_1"
+                       } = stu_1
+                     ]
+                   }
+                 }
+               ]
+             } = encoded
+
+      refute Map.get(stu_1, "arrival")
+      refute Map.get(stu_1, "departure")
+      refute Map.get(stu_2, "arrival")
+      refute Map.get(stu_2, "departure")
+      refute Map.get(stu_3, "passthrough_time")
+    end
   end
 end

test/concentrate/encoder/trip_updates_test.exs

@@ -178,5 +178,29 @@ defmodule Concentrate.Encoder.TripUpdatesTest do
                entity: []
              } = decoded
     end
+
+    test "stop time updates with a passthrough_time are removed" do
+      initial = [
+        TripDescriptor.new(trip_id: "1"),
+        StopTimeUpdate.new(trip_id: "1", stop_sequence: 1, departure_time: 1),
+        StopTimeUpdate.new(trip_id: "1", stop_sequence: 2, arrival_time: 2, passthrough_time: 2),
+        StopTimeUpdate.new(trip_id: "1", stop_sequence: 3, arrival_time: 3, departure_time: 4),
+        StopTimeUpdate.new(trip_id: "1", stop_sequence: 4, departure_time: 5, passthrough_time: 5)
+      ]
+
+      decoded = GTFSRealtime.parse(encode_groups(group(initial)), [])
+
+      assert [
+               TripDescriptor.new(trip_id: "1"),
+               StopTimeUpdate.new(trip_id: "1", stop_sequence: 1, departure_time: 1),
+               StopTimeUpdate.new(
+                 trip_id: "1",
+                 stop_sequence: 3,
+                 arrival_time: 3,
+                 departure_time: 4
+               )
+             ] ==
+               FeedUpdate.updates(decoded)
+    end
   end
 end

test/concentrate/group_filter/skipped_stop_on_added_trip_test.exs

@@ -31,6 +31,32 @@ defmodule Concentrate.Filter.SkippedStopOnAddedTripTest do
       assert {^td, [], [^stu]} = filter({td, [], [stu]})
     end
 
+    test "keeps stus with passthough_times from ADDED trips" do
+      td = TripDescriptor.new(trip_id: @trip_id, schedule_relationship: :ADDED)
+
+      stu =
+        StopTimeUpdate.new(
+          trip_id: @trip_id,
+          schedule_relationship: :SKIPPED,
+          passthrough_time: 500
+        )
+
+      assert {^td, [], [^stu]} = filter({td, [], [stu]})
+    end
+
+    test "keeps stus with passthough_times from UNSCHEDULED trips" do
+      td = TripDescriptor.new(trip_id: @trip_id, schedule_relationship: :UNSCHEDULED)
+
+      stu =
+        StopTimeUpdate.new(
+          trip_id: @trip_id,
+          schedule_relationship: :SKIPPED,
+          passthrough_time: 500
+        )
+
+      assert {^td, [], [^stu]} = filter({td, [], [stu]})
+    end
+
     test "other values are returned as-is" do
       assert filter(:value) == :value
     end

test/concentrate/parser/gtfs_realtime_enhanced_test.exs

@@ -411,6 +411,30 @@ defmodule Concentrate.Parser.GTFSRealtimeEnhancedTest do
       assert stu.uncertainty == 500
     end
 
+    test "parses passthrough_time" do
+      update = %{
+        "trip" => %{
+          "trip_id" => "trip",
+          "route_id" => "route"
+        },
+        "stop_time_update" => [
+          %{
+            "arrival" => %{"time" => 100, "uncertainty" => 500},
+            "departure" => %{"time" => 200, "uncertainty" => 500},
+            "passthrough_time" => 100
+          },
+          %{
+            "arrival" => %{"time" => 300, "uncertainty" => 500},
+            "departure" => %{"time" => 400, "uncertainty" => 500}
+          }
+        ]
+      }
+
+      [_td, stu1, stu2] = decode_trip_update(update, %Options{})
+      assert stu1.passthrough_time == 100
+      refute stu2.passthrough_time
+    end
+
     test "decodes last_trip" do
       not_last_trip = %{
         "trip" => %{

test/concentrate/stop_time_update_test.exs

@@ -51,7 +51,8 @@ defmodule Concentrate.StopTimeUpdateTest do
           departure_time: 2,
           track: "track",
           schedule_relationship: :SKIPPED,
-          uncertainty: 300
+          uncertainty: 300,
+          passthrough_time: 1
         )
 
       expected =
@@ -65,7 +66,8 @@ defmodule Concentrate.StopTimeUpdateTest do
           track: "track",
           schedule_relationship: :SKIPPED,
           platform_id: "platform",
-          uncertainty: 300
+          uncertainty: 300,
+          passthrough_time: 1
         )
 
       assert Mergeable.merge(first, second) == expected