Revert "Migrate to GTFS-rt predictions and locations feeds (#743)"

This reverts commit 58bf0b6.

.envrc.template

@@ -17,6 +17,7 @@ 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. 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"
+# 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=

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/concentrate/TripUpdates_enhanced.json"
+        "https://s3.amazonaws.com/mbta-gtfs-s3/rtr/TripUpdates_enhanced.json"
       ),
     vehicle_positions_url:
       System.get_env(
         "VEHICLE_POSITIONS_URL",
-        "https://s3.amazonaws.com/mbta-gtfs-s3/concentrate/VehiclePositions_enhanced.json"
+        "https://s3.amazonaws.com/mbta-gtfs-s3/rtr/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,17 +77,10 @@ defmodule Content.Message.Predictions do
 
     {minutes, approximate?} =
       cond do
-        prediction.stopped_at_predicted_stop? ->
-          {:boarding, false}
-
-        predicted_time <= 30 ->
-          {:arriving, false}
-
-        predicted_time <= 60 ->
-          {:approaching, false}
-
-        true ->
-          compute_minutes(predicted_time, certainty)
+        prediction.stops_away == 0 -> {:boarding, false}
+        predicted_time <= 30 -> {:arriving, false}
+        predicted_time <= 60 -> {:approaching, false}
+        true -> compute_minutes(predicted_time, certainty)
       end
 
     {crowding_data_confidence, crowding_description} =
@@ -135,16 +128,13 @@ 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 prediction.stopped_at_predicted_stop? ->
-          {:boarding, false}
-
-        x when x <= @terminal_brd_seconds ->
-          {1, false}
-
-        x ->
-          compute_minutes(x, prediction.departure_certainty)
+        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)
       end
 
     case Content.Utilities.destination_for_prediction(

lib/fake/httpoison.ex

@@ -65,11 +65,15 @@ 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
                   },
@@ -80,6 +84,7 @@ defmodule Fake.HTTPoison do
                 },
                 %{
                   "arrival" => %{
+                    "delay" => nil,
                     "time" => 1_491_570_180,
                     "uncertainty" => nil
                   },
@@ -238,17 +243,23 @@ 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" => 60
+                    "uncertainty" => nil
                   },
                   "departure" => nil,
+                  "schedule_relationship" => "SCHEDULED",
                   "stop_id" => "stop_to_update",
-                  "stop_sequence" => 1
+                  "stop_sequence" => 1,
+                  "stops_away" => 0
                 }
               ],
               "timestamp" => nil,
@@ -258,15 +269,15 @@ defmodule Fake.HTTPoison do
                 "schedule_relationship" => "SCHEDULED",
                 "start_date" => "20170329",
                 "start_time" => nil,
-                "trip_id" => "32568935",
-                "revenue" => true
+                "trip_id" => "32568935"
               },
               "vehicle" => %{
                 "id" => "G-10040",
                 "label" => "3260",
                 "license_plate" => nil
               }
-            }
+            },
+            "vehicle" => nil
           }
         ],
         "header" => %{

lib/predictions/last_trip.ex

@@ -1,13 +1,8 @@
 defmodule Predictions.LastTrip do
-  alias Predictions.Predictions
-
   defp get_running_trips(predictions_feed) do
     predictions_feed["entity"]
     |> Stream.map(& &1["trip_update"])
-    |> Stream.filter(
-      &(Predictions.relevant_rail_route?(&1["trip"]["route_id"]) and
-          &1["trip"]["schedule_relationship"] != "CANCELED")
-    )
+    |> Enum.reject(&(&1["trip"]["schedule_relationship"] == "CANCELED"))
   end
 
   def get_last_trips(predictions_feed) do

lib/predictions/prediction.ex

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

lib/predictions/predictions.ex

@@ -9,27 +9,23 @@ defmodule Predictions.Predictions do
   def get_all(feed_message, current_time) do
     predictions =
       feed_message["entity"]
-      |> Stream.map(& &1["trip_update"])
-      |> Stream.filter(
-        &(relevant_rail_route?(&1["trip"]["route_id"]) 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"])
+      |> 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"]
       end)
-      |> Stream.map(&prediction_from_update(&1, current_time))
+      |> Enum.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)) or
-            (&1.seconds_until_departure && &1.seconds_until_departure < 0))
+        &(is_nil(&1.seconds_until_arrival) and is_nil(&1.seconds_until_departure) and
+            is_nil(&1.seconds_until_passthrough))
       )
 
     vehicles_running_revenue_trips =
       predictions
-      |> Stream.filter(& &1.revenue_trip?)
-      |> Stream.map(& &1.vehicle_id)
+      |> Enum.filter(& &1.revenue_trip?)
+      |> Enum.map(& &1.vehicle_id)
       |> MapSet.new()
 
     {Enum.group_by(predictions, fn prediction ->
@@ -47,12 +43,15 @@ 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"], trip_update["trip"]["revenue"], vehicle_id}
+       trip_update["trip"]["trip_id"], revenue_trip?, vehicle_id}
     )
   end
 
@@ -81,28 +80,25 @@ defmodule Predictions.Predictions do
          else: nil
 
     seconds_until_passthrough =
-      if not revenue_trip?,
-        do: seconds_until_arrival || seconds_until_departure,
+      if stop_time_update["passthrough_time"],
+        do: stop_time_update["passthrough_time"] - current_time_seconds,
         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: seconds_until_departure,
+      seconds_until_departure: max(0, 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_at_predicted_stop?:
-        not is_nil(vehicle_location) and vehicle_location.status == :stopped_at and
-          stop_time_update["stop_id"] == vehicle_location.stop_id,
+      stopped?: stop_time_update["stopped?"],
+      stops_away: stop_time_update["stops_away"],
       boarding_status: stop_time_update["boarding_status"],
       revenue_trip?: revenue_trip?,
       vehicle_id: vehicle_id
@@ -117,19 +113,6 @@ defmodule Predictions.Predictions do
     Jason.decode!(body)
   end
 
-  def relevant_rail_route?(route_id) do
-    route_id in [
-      "Red",
-      "Blue",
-      "Orange",
-      "Green-B",
-      "Green-C",
-      "Green-D",
-      "Green-E",
-      "Mattapan"
-    ]
-  end
-
   @spec translate_schedule_relationship(String.t()) :: :skipped | :scheduled
   defp translate_schedule_relationship("SKIPPED") do
     :skipped

lib/signs/utilities/predictions.ex

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