🧠 Implement "uncertainty adapter" within Concentrate

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.GropuFilter.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,29 @@
+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}) do
+    stus = set_uncertainty(update_type, stus)
+
+    {td, vps, stus}
+  end
+
+  def filter(other), do: other
+
+  defp set_uncertainty(nil, stus), do: stus
+
+  defp set_uncertainty(update_type, stus) do
+    Enum.map(stus, fn stu ->
+      StopTimeUpdate.update_uncertainty(stu, calculate_uncertainty(update_type))
+    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

@@ -131,8 +131,8 @@ defmodule Concentrate.Parser.GTFSRealtimeEnhanced do
        ) do
     max_time = options.max_time
 
-    {arrival_time, _} = time_from_event(Map.get(update, "arrival"))
-    {departure_time, _} = time_from_event(Map.get(update, "departure"))
+    arrival_time = time_from_event(Map.get(update, "arrival"))
+    departure_time = time_from_event(Map.get(update, "departure"))
 
     cond do
       td != [] and not Helpers.valid_route_id?(options, TripDescriptor.route_id(List.first(td))) ->
@@ -144,8 +144,9 @@ defmodule Concentrate.Parser.GTFSRealtimeEnhanced do
       true ->
         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"))
+            arrival_time = time_from_event(Map.get(stu, "arrival"))
+
+            departure_time = time_from_event(Map.get(stu, "departure"))
 
             boarding_status = decode_boarding_status(Map.get(stu, "boarding_status"))
 
@@ -157,7 +158,6 @@ defmodule Concentrate.Parser.GTFSRealtimeEnhanced do
               schedule_relationship: schedule_relationship(Map.get(stu, "schedule_relationship")),
               arrival_time: arrival_time,
               departure_time: departure_time,
-              uncertainty: arrival_uncertainty || departure_uncertainty,
               status: boarding_status,
               platform_id: Map.get(stu, "platform_id")
             )
@@ -285,8 +285,9 @@ defmodule Concentrate.Parser.GTFSRealtimeEnhanced do
     Date.to_erl(date)
   end
 
-  defp time_from_event(nil), do: {nil, nil}
-  defp time_from_event(%{"time" => time} = map), do: {time, Map.get(map, "uncertainty", nil)}
+  defp time_from_event(nil), do: nil
+
+  defp time_from_event(%{"time" => time}), do: time
 
   defp schedule_relationship(nil), do: :SCHEDULED
 

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 TripDescriptor 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 TripDescriptor 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 TripDescriptor 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 "populates uncertainty in TripDescriptor based on update_type value of other" do
+      td = TripDescriptor.new(update_type: nil)
+
+      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) == nil
+             end)
+    end
+  end
+end

test/concentrate/parser/gtfs_realtime_enhanced_test.exs

@@ -372,6 +372,24 @@ defmodule Concentrate.Parser.GTFSRealtimeEnhancedTest do
       assert TripDescriptor.revenue(td) == true
     end
 
+    test "update_type is nil 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{})
+      assert stu.uncertainty == nil
+    end
+
     test "decodes last_trip" do
       not_last_trip = %{
         "trip" => %{