🧠 Implement "uncertainty adapter" within Concentrate (#346)

config/config.exs

@@ -73,7 +73,8 @@ config :concentrate,
     Concentrate.GroupFilter.VehicleAtSkippedStop,
     Concentrate.GroupFilter.VehicleStopMatch,
     Concentrate.GroupFilter.SkippedStopOnAddedTrip,
-    Concentrate.GroupFilter.TripDescriptorTimestamp
+    Concentrate.GroupFilter.TripDescriptorTimestamp,
+    Concentrate.GroupFilter.UncertaintyValue
   ],
   source_reporters: [
     Concentrate.SourceReporter.Basic,

lib/concentrate/encoder/gtfs_realtime_helpers.ex

@@ -203,6 +203,8 @@ defmodule Concentrate.Encoder.GTFSRealtimeHelpers do
       |> Map.merge(enhanced_data_fn.(td))
       |> drop_nil_values()
 
+    {update_type, trip} = Map.pop(trip, :update_type)
+
     vehicle = trip_update_vehicle(td, vps)
 
     stop_time_update =
@@ -221,7 +223,8 @@ defmodule Concentrate.Encoder.GTFSRealtimeHelpers do
                 trip: trip,
                 stop_time_update: stop_time_update,
                 vehicle: vehicle,
-                timestamp: timestamp
+                timestamp: timestamp,
+                update_type: update_type
               })
           }
         ]
@@ -230,7 +233,13 @@ defmodule Concentrate.Encoder.GTFSRealtimeHelpers do
         [
           %{
             id: id,
-            trip_update: drop_nil_values(%{trip: trip, vehicle: vehicle, timestamp: timestamp})
+            trip_update:
+              drop_nil_values(%{
+                trip: trip,
+                vehicle: vehicle,
+                timestamp: timestamp,
+                update_type: update_type
+              })
           }
         ]
 

lib/concentrate/encoder/trip_updates_enhanced.ex

@@ -20,7 +20,8 @@ defmodule Concentrate.Encoder.TripUpdatesEnhanced do
     %{
       route_pattern_id: TripDescriptor.route_pattern_id(update),
       revenue: TripDescriptor.revenue(update),
-      last_trip: TripDescriptor.last_trip(update)
+      last_trip: TripDescriptor.last_trip(update),
+      update_type: TripDescriptor.update_type(update)
     }
   end
 

lib/concentrate/group_filter/uncertainty_value.ex

@@ -0,0 +1,33 @@
+defmodule Concentrate.GroupFilter.UncertaintyValue do
+  @moduledoc """
+  Populates uncertainty in TripDescriptor based on the update_type value
+  """
+  @behaviour Concentrate.GroupFilter
+  alias Concentrate.{StopTimeUpdate, TripDescriptor}
+
+  @impl Concentrate.GroupFilter
+  def filter({%TripDescriptor{update_type: update_type} = td, vps, stus})
+      when not is_nil(update_type) do
+    stus =
+      update_type
+      |> calculate_uncertainty()
+      |> set_uncertainty(stus)
+
+    {td, vps, stus}
+  end
+
+  def filter(other), do: other
+
+  defp set_uncertainty(nil, stus), do: stus
+
+  defp set_uncertainty(uncertainty, stus) do
+    Enum.map(stus, fn stu ->
+      StopTimeUpdate.update_uncertainty(stu, uncertainty)
+    end)
+  end
+
+  defp calculate_uncertainty("mid_trip"), do: 60
+  defp calculate_uncertainty("at_terminal"), do: 120
+  defp calculate_uncertainty("reverse_trip"), do: 360
+  defp calculate_uncertainty(_), do: nil
+end

lib/concentrate/parser/gtfs_realtime_enhanced.ex

@@ -145,6 +145,7 @@ defmodule Concentrate.Parser.GTFSRealtimeEnhanced do
         stop_updates =
           for stu <- updates do
             {arrival_time, arrival_uncertainty} = time_from_event(Map.get(stu, "arrival"))
+
             {departure_time, departure_uncertainty} = time_from_event(Map.get(stu, "departure"))
 
             boarding_status = decode_boarding_status(Map.get(stu, "boarding_status"))
@@ -249,7 +250,8 @@ defmodule Concentrate.Parser.GTFSRealtimeEnhanced do
         timestamp: Map.get(descriptor, "timestamp"),
         revenue: Map.get(trip, "revenue", true),
         vehicle_id: vehicle_id,
-        last_trip: Map.get(trip, "last_trip", false)
+        last_trip: Map.get(trip, "last_trip", false),
+        update_type: Map.get(descriptor, "update_type")
       )
     ]
   end

lib/concentrate/trip_descriptor.ex

@@ -13,6 +13,7 @@ defmodule Concentrate.TripDescriptor do
     :start_time,
     :vehicle_id,
     :timestamp,
+    :update_type,
     last_trip: false,
     revenue: true,
     schedule_relationship: :SCHEDULED
@@ -59,13 +60,17 @@ defmodule Concentrate.TripDescriptor do
           vehicle_id: first.vehicle_id || second.vehicle_id,
           timestamp: first.timestamp || second.timestamp,
           last_trip: first.last_trip || second.last_trip,
-          schedule_relationship: merge_schedule_relationship(first, second)
+          schedule_relationship: merge_schedule_relationship(first, second),
+          update_type: merge_update_type(first.update_type, second.update_type)
       }
     end
 
     defp merge_schedule_relationship(%{schedule_relationship: :SCHEDULED}, second),
       do: second.schedule_relationship
 
     defp merge_schedule_relationship(first, _), do: first.schedule_relationship
+
+    defp merge_update_type(nil, second), do: second
+    defp merge_update_type(first, _), do: first
   end
 end

test/concentrate/encoder/trip_updates_enhanced_test.exs

@@ -196,5 +196,28 @@ defmodule Concentrate.Encoder.TripUpdatesEnhancedTest do
                ]
              } = encoded
     end
+
+    test "moves update_type from trip to trip_update" do
+      parsed = [
+        TripDescriptor.new(trip_id: "trip", update_type: "mid_trip"),
+        StopTimeUpdate.new(trip_id: "trip", stop_id: "stop")
+      ]
+
+      encoded = Jason.decode!(encode_groups(group(parsed)))
+
+      assert %{
+               "entity" => [
+                 %{
+                   "trip_update" => %{
+                     "update_type" => "mid_trip",
+                     "trip" => %{}
+                   }
+                 }
+               ]
+             } = encoded
+
+      trip = get_in(encoded, ["entity", Access.at(0), "trip_update", "trip"])
+      refute trip["update_type"]
+    end
   end
 end

test/concentrate/group_filter/unceratinty_value_test.exs

@@ -0,0 +1,73 @@
+defmodule Concentrate.GroupFilter.UncertaintyValueTest do
+  @moduledoc false
+  use ExUnit.Case, async: true
+  import Concentrate.GroupFilter.UncertaintyValue
+  alias Concentrate.StopTimeUpdate
+  alias Concentrate.TripDescriptor
+
+  describe "filter/1" do
+    test "populates uncertainty in StopTimeUpdate based on update_type value of mid_trip" do
+      td = TripDescriptor.new(update_type: "mid_trip")
+
+      stus = [
+        StopTimeUpdate.new(uncertainty: nil),
+        StopTimeUpdate.new(uncertainty: nil),
+        StopTimeUpdate.new(uncertainty: nil)
+      ]
+
+      {^td, [], processed_stus} = filter({td, [], stus})
+
+      assert Enum.all?(processed_stus, fn procced_stu ->
+               StopTimeUpdate.uncertainty(procced_stu) == 60
+             end)
+    end
+
+    test "populates uncertainty in StopTimeUpdate based on update_type value of at_terminal" do
+      td = TripDescriptor.new(update_type: "at_terminal")
+
+      stus = [
+        StopTimeUpdate.new(uncertainty: nil),
+        StopTimeUpdate.new(uncertainty: nil),
+        StopTimeUpdate.new(uncertainty: nil)
+      ]
+
+      {^td, [], processed_stus} = filter({td, [], stus})
+
+      assert Enum.all?(processed_stus, fn procced_stu ->
+               StopTimeUpdate.uncertainty(procced_stu) == 120
+             end)
+    end
+
+    test "populates uncertainty in StopTimeUpdate based on update_type value of reverse_trip" do
+      td = TripDescriptor.new(update_type: "reverse_trip")
+
+      stus = [
+        StopTimeUpdate.new(uncertainty: nil),
+        StopTimeUpdate.new(uncertainty: nil),
+        StopTimeUpdate.new(uncertainty: nil)
+      ]
+
+      {^td, [], processed_stus} = filter({td, [], stus})
+
+      assert Enum.all?(processed_stus, fn procced_stu ->
+               StopTimeUpdate.uncertainty(procced_stu) == 360
+             end)
+    end
+
+    test "maintains uncertainty in StopTimeUpdate if update_type is not valid" do
+      td = TripDescriptor.new(update_type: nil)
+
+      stus = [
+        StopTimeUpdate.new(uncertainty: 60),
+        StopTimeUpdate.new(uncertainty: 60),
+        StopTimeUpdate.new(uncertainty: 60)
+      ]
+
+      {^td, [], processed_stus} = filter({td, [], stus})
+
+      assert Enum.all?(processed_stus, fn procced_stu ->
+               StopTimeUpdate.uncertainty(procced_stu) == 60
+             end)
+    end
+  end
+end

test/concentrate/parser/gtfs_realtime_enhanced_test.exs

@@ -372,6 +372,45 @@ defmodule Concentrate.Parser.GTFSRealtimeEnhancedTest do
       assert TripDescriptor.revenue(td) == true
     end
 
+    test "update_type and uncertainty values are set if present" do
+      update = %{
+        "trip" => %{
+          "trip_id" => "trip",
+          "route_id" => "route"
+        },
+        "update_type" => "mid_trip",
+        "stop_time_update" => [
+          %{
+            "arrival" => %{"time" => 100, "uncertainty" => 500},
+            "departure" => %{"time" => 200, "uncertainty" => 500}
+          }
+        ]
+      }
+
+      [td, stu] = decode_trip_update(update, %Options{})
+      assert td.update_type == "mid_trip"
+      assert stu.uncertainty == 500
+    end
+
+    test "uncertainty uses uncertainty value if update_type is not present" do
+      update = %{
+        "trip" => %{
+          "trip_id" => "trip",
+          "route_id" => "route"
+        },
+        "stop_time_update" => [
+          %{
+            "arrival" => %{"time" => 100, "uncertainty" => 500},
+            "departure" => %{"time" => 200, "uncertainty" => 500}
+          }
+        ]
+      }
+
+      [td, stu] = decode_trip_update(update, %Options{})
+      refute td.update_type
+      assert stu.uncertainty == 500
+    end
+
     test "decodes last_trip" do
       not_last_trip = %{
         "trip" => %{