Migrate to GTFS-rt predictions and locations feeds

.envrc.template

@@ -17,7 +17,6 @@ export API_V3_URL=https://api-dev-green.mbtace.com
 #export CHELSEA_BRIDGE_URL=
 #export CHELSEA_BRIDGE_AUTH=
 
-# URLs of the enhanced trip-update and vehicle-position feeds. Default to the real feed URLs if
-# not set here.
-#export TRIP_UPDATE_URL=
-#export VEHICLE_POSITIONS_URL=
+# URLs of the enhanced trip-update and vehicle-position feeds. Required.
+#export TRIP_UPDATE_URL="https://s3.amazonaws.com/mbta-gtfs-s3/concentrate/TripUpdates_enhanced.json"
+#export VEHICLE_POSITIONS_URL="https://s3.amazonaws.com/mbta-gtfs-s3/concentrate/VehiclePositions_enhanced.json"

config/runtime.exs

@@ -12,12 +12,12 @@ if config_env() != :test do
     trip_update_url:
       System.get_env(
         "TRIP_UPDATE_URL",
-        "https://s3.amazonaws.com/mbta-gtfs-s3/rtr/TripUpdates_enhanced.json"
+        "https://s3.amazonaws.com/mbta-gtfs-s3/concentrate/TripUpdates_enhanced.json"
       ),
     vehicle_positions_url:
       System.get_env(
         "VEHICLE_POSITIONS_URL",
-        "https://s3.amazonaws.com/mbta-gtfs-s3/rtr/VehiclePositions_enhanced.json"
+        "https://s3.amazonaws.com/mbta-gtfs-s3/concentrate/VehiclePositions_enhanced.json"
       ),
     s3_bucket: System.get_env("SIGNS_S3_BUCKET"),
     s3_path: System.get_env("SIGNS_S3_PATH"),

lib/content/message/predictions.ex

@@ -77,10 +77,17 @@ defmodule Content.Message.Predictions do
 
     {minutes, approximate?} =
       cond do
-        prediction.stops_away == 0 -> {:boarding, false}
-        predicted_time <= 30 -> {:arriving, false}
-        predicted_time <= 60 -> {:approaching, false}
-        true -> compute_minutes(predicted_time, certainty)
+        prediction.stopped_at_predicted_stop? ->
+          {:boarding, false}
+
+        predicted_time <= 30 ->
+          {:arriving, false}
+
+        predicted_time <= 60 ->
+          {:approaching, false}
+
+        true ->
+          compute_minutes(predicted_time, certainty)
       end
 
     {crowding_data_confidence, crowding_description} =
@@ -128,13 +135,16 @@ defmodule Content.Message.Predictions do
   def terminal(prediction, station_code, zone, sign, width \\ 18)
 
   def terminal(prediction, station_code, zone, sign, width) do
-    stopped_at? = prediction.stops_away == 0
-
     {minutes, approximate?} =
       case prediction.seconds_until_departure + @terminal_prediction_offset_seconds do
-        x when x <= @terminal_brd_seconds and stopped_at? -> {:boarding, false}
-        x when x <= @terminal_brd_seconds -> {1, false}
-        x -> compute_minutes(x, prediction.departure_certainty)
+        x when x <= @terminal_brd_seconds and prediction.stopped_at_predicted_stop? ->
+          {:boarding, false}
+
+        x when x <= @terminal_brd_seconds ->
+          {1, false}
+
+        x ->
+          compute_minutes(x, prediction.departure_certainty)
       end
 
     case Content.Utilities.destination_for_prediction(

lib/fake/httpoison.ex

@@ -65,15 +65,11 @@ defmodule Fake.HTTPoison do
       %{
         "entity" => [
           %{
-            "alert" => nil,
             "id" => "1490783458_32568935",
-            "is_deleted" => false,
             "trip_update" => %{
-              "delay" => nil,
               "stop_time_update" => [
                 %{
                   "arrival" => %{
-                    "delay" => nil,
                     "time" => 1_491_570_120,
                     "uncertainty" => nil
                   },
@@ -84,7 +80,6 @@ defmodule Fake.HTTPoison do
                 },
                 %{
                   "arrival" => %{
-                    "delay" => nil,
                     "time" => 1_491_570_180,
                     "uncertainty" => nil
                   },
@@ -133,23 +128,17 @@ defmodule Fake.HTTPoison do
       %{
         "entity" => [
           %{
-            "alert" => nil,
             "id" => "1490783458_32568935",
-            "is_deleted" => false,
             "trip_update" => %{
-              "delay" => nil,
               "stop_time_update" => [
                 %{
                   "arrival" => %{
-                    "delay" => nil,
                     "time" => 1_491_570_180,
-                    "uncertainty" => nil
+                    "uncertainty" => 60
                   },
                   "departure" => nil,
-                  "schedule_relationship" => "SCHEDULED",
                   "stop_id" => "stop_to_update",
-                  "stop_sequence" => 1,
-                  "stops_away" => 0
+                  "stop_sequence" => 1
                 }
               ],
               "timestamp" => nil,
@@ -159,15 +148,15 @@ defmodule Fake.HTTPoison do
                 "schedule_relationship" => "SCHEDULED",
                 "start_date" => "20170329",
                 "start_time" => nil,
-                "trip_id" => "32568935"
+                "trip_id" => "32568935",
+                "revenue" => true
               },
               "vehicle" => %{
                 "id" => "G-10040",
                 "label" => "3260",
                 "license_plate" => nil
               }
-            },
-            "vehicle" => nil
+            }
           }
         ],
         "header" => %{

lib/predictions/last_trip.ex

@@ -2,7 +2,18 @@ defmodule Predictions.LastTrip do
   defp get_running_trips(predictions_feed) do
     predictions_feed["entity"]
     |> Stream.map(& &1["trip_update"])
-    |> Enum.reject(&(&1["trip"]["schedule_relationship"] == "CANCELED"))
+    |> Stream.filter(
+      &(&1["trip"]["route_id"] in [
+          "Red",
+          "Blue",
+          "Orange",
+          "Green-B",
+          "Green-C",
+          "Green-D",
+          "Green-E",
+          "Mattapan"
+        ] and &1["trip"]["schedule_relationship"] != "CANCELED")
+    )
   end
 
   def get_last_trips(predictions_feed) do

lib/predictions/prediction.ex

@@ -10,8 +10,7 @@ defmodule Predictions.Prediction do
             route_id: nil,
             trip_id: nil,
             destination_stop_id: nil,
-            stopped?: false,
-            stops_away: 0,
+            stopped_at_predicted_stop?: false,
             boarding_status: nil,
             revenue_trip?: true,
             vehicle_id: nil
@@ -30,8 +29,7 @@ defmodule Predictions.Prediction do
           route_id: String.t(),
           trip_id: trip_id() | nil,
           destination_stop_id: String.t(),
-          stopped?: boolean(),
-          stops_away: integer(),
+          stopped_at_predicted_stop?: boolean(),
           boarding_status: String.t() | nil,
           revenue_trip?: boolean(),
           vehicle_id: String.t() | nil

lib/predictions/predictions.ex

@@ -9,23 +9,35 @@ defmodule Predictions.Predictions do
   def get_all(feed_message, current_time) do
     predictions =
       feed_message["entity"]
-      |> Enum.map(& &1["trip_update"])
-      |> Enum.reject(&(&1["trip"]["schedule_relationship"] == "CANCELED"))
-      |> Enum.flat_map(&transform_stop_time_updates/1)
-      |> Enum.filter(fn {update, _, _, _, _, _, _} ->
-        ((update["arrival"] || update["departure"]) &&
-           not is_nil(update["stops_away"])) || update["passthrough_time"]
+      |> Stream.map(& &1["trip_update"])
+      |> Stream.filter(
+        &(&1["trip"]["route_id"] in [
+            "Red",
+            "Blue",
+            "Orange",
+            "Green-B",
+            "Green-C",
+            "Green-D",
+            "Green-E",
+            "Mattapan"
+          ] and &1["trip"]["schedule_relationship"] != "CANCELED")
+      )
+      |> Stream.flat_map(&transform_stop_time_updates/1)
+      |> Stream.filter(fn {update, _, _, _, _, _, _} ->
+        (update["arrival"] && update["arrival"]["uncertainty"]) ||
+          (update["departure"] && update["departure"]["uncertainty"])
       end)
-      |> Enum.map(&prediction_from_update(&1, current_time))
+      |> Stream.map(&prediction_from_update(&1, current_time))
       |> Enum.reject(
-        &(is_nil(&1.seconds_until_arrival) and is_nil(&1.seconds_until_departure) and
-            is_nil(&1.seconds_until_passthrough))
+        &((is_nil(&1.seconds_until_arrival) and is_nil(&1.seconds_until_departure) and
+             is_nil(&1.seconds_until_passthrough)) or
+            (&1.seconds_until_departure && &1.seconds_until_departure < 0))
       )
 
     vehicles_running_revenue_trips =
       predictions
-      |> Enum.filter(& &1.revenue_trip?)
-      |> Enum.map(& &1.vehicle_id)
+      |> Stream.filter(& &1.revenue_trip?)
+      |> Stream.map(& &1.vehicle_id)
       |> MapSet.new()
 
     {Enum.group_by(predictions, fn prediction ->
@@ -43,15 +55,12 @@ defmodule Predictions.Predictions do
       end)
       |> Map.get("stop_id")
 
-    revenue_trip? =
-      Enum.any?(trip_update["stop_time_update"], &(&1["schedule_relationship"] != "SKIPPED"))
-
     vehicle_id = get_in(trip_update, ["vehicle", "id"])
 
     Enum.map(
       trip_update["stop_time_update"],
       &{&1, last_stop_id, trip_update["trip"]["route_id"], trip_update["trip"]["direction_id"],
-       trip_update["trip"]["trip_id"], revenue_trip?, vehicle_id}
+       trip_update["trip"]["trip_id"], trip_update["trip"]["revenue"], vehicle_id}
     )
   end
 
@@ -80,25 +89,28 @@ defmodule Predictions.Predictions do
          else: nil
 
     seconds_until_passthrough =
-      if stop_time_update["passthrough_time"],
-        do: stop_time_update["passthrough_time"] - current_time_seconds,
+      if not revenue_trip?,
+        do: seconds_until_arrival || seconds_until_departure,
         else: nil
 
+    vehicle_location = Engine.Locations.for_vehicle(vehicle_id)
+
     %Prediction{
       stop_id: stop_time_update["stop_id"],
       direction_id: direction_id,
       seconds_until_arrival: max(0, seconds_until_arrival),
       arrival_certainty: stop_time_update["arrival"]["uncertainty"],
-      seconds_until_departure: max(0, seconds_until_departure),
+      seconds_until_departure: seconds_until_departure,
       departure_certainty: stop_time_update["departure"]["uncertainty"],
       seconds_until_passthrough: max(0, seconds_until_passthrough),
       schedule_relationship:
         translate_schedule_relationship(stop_time_update["schedule_relationship"]),
       route_id: route_id,
       trip_id: trip_id,
       destination_stop_id: last_stop_id,
-      stopped?: stop_time_update["stopped?"],
-      stops_away: stop_time_update["stops_away"],
+      stopped_at_predicted_stop?:
+        not is_nil(vehicle_location) and vehicle_location.status == :stopped_at and
+          stop_time_update["stop_id"] == vehicle_location.stop_id,
       boarding_status: stop_time_update["boarding_status"],
       revenue_trip?: revenue_trip?,
       vehicle_id: vehicle_id

lib/signs/utilities/predictions.ex

@@ -47,10 +47,9 @@ defmodule Signs.Utilities.Predictions do
       {if terminal_prediction?(prediction, sources) do
          0
        else
-         case prediction.stops_away do
-           0 -> 0
-           _ -> 1
-         end
+         if prediction.stopped_at_predicted_stop?,
+           do: 0,
+           else: 1
        end, prediction.seconds_until_departure, prediction.seconds_until_arrival}
     end)
     |> filter_large_red_line_gaps()