feat(predictions): add schedule relationship filter (#1087)

Summary of changes

Asana Ticket: 🔮 Build a basic Splunk dashboard for RTR

Problem:
As part of building out observability for our subway predictions, we need to periodically check how many ADDED / SKIPPED predictions we are making for each line and direction. We want to add this check in the api_checker, which polls the V3 API. However, no filter exists for schedule_relationship on the predictions endpoint.

Solution:
Add a filter for schedule_relationship on the predictions endpoint. Following this, we can add product health checks that look at the proportion of ADDED / SKIPPED predictions we are making per route / direction.

Author: rudiejd

apps/api_web/lib/api_web/controllers/prediction_controller.ex

@@ -7,13 +7,19 @@ defmodule ApiWeb.PredictionController do
   * route
   * trip
   * radius
+  * direction_id
+  * stop_sequence
+  * route_type
+  * route_pattern
+  * revenue
+  * schedule_relationship
   """
   use ApiWeb.Web, :api_controller
   require Logger
   alias ApiWeb.LegacyStops
   alias State.Prediction
 
-  @filters ~w(stop route trip latitude longitude radius direction_id stop_sequence route_type route_pattern revenue)s
+  @filters ~w(stop route trip latitude longitude radius direction_id stop_sequence route_type route_pattern revenue schedule_relationship)s
   @pagination_opts ~w(offset limit order_by)a
   @includes ~w(schedule stop route trip vehicle alerts)
 
@@ -63,6 +69,11 @@ defmodule ApiWeb.PredictionController do
     filter_param(:id, name: :trip)
     filter_param(:revenue, desc: "Filter predictions by revenue status.")
 
+    filter_param(:schedule_relationship,
+      desc:
+        "Filter predictions by schedule relationship: https://github.com/google/transit/blob/master/gtfs-realtime/spec/en/reference.md#enum-schedulerelationship"
+    )
+
     parameter("filter[route_pattern]", :query, :string, """
     Filter by `/included/{index}/relationships/route_pattern/data/id` of a trip. Multiple `route_pattern_id` #{comma_separated_list()}.
     """)
@@ -79,9 +90,15 @@ defmodule ApiWeb.PredictionController do
     with :ok <- Params.validate_includes(params, @includes, conn),
          {:ok, filtered_params} <- Params.filter_params(params, filters(conn), conn) do
       case filtered_params do
+        %{"schedule_relationship" => _} = p when map_size(p) == 1 ->
+          {:error, :only_schedule_relationship}
+
         %{"route_type" => _} = p when map_size(p) == 1 ->
           {:error, :only_route_type}
 
+        %{"route_type" => _, "schedule_relationship" => _} = p when map_size(p) == 2 ->
+          {:error, :only_route_type_and_schedule_relationship}
+
         p when map_size(p) > 0 ->
           do_index_data(conn, params, filtered_params)
 
@@ -101,6 +118,9 @@ defmodule ApiWeb.PredictionController do
     route_ids = Params.split_on_comma(filtered_params, "route")
     route_types = Params.route_types(filtered_params)
 
+    schedule_relationships =
+      Params.schedule_relationships(filtered_params)
+
     pagination_opts =
       Params.filter_opts(params, @pagination_opts, conn, order_by: {:arrival_time, :asc})
 
@@ -113,6 +133,7 @@ defmodule ApiWeb.PredictionController do
       filtered_params
       |> build_stop_sequence_matchers(direction_id_matcher)
       |> add_revenue_matchers(revenue)
+      |> add_schedule_relationship_matchers(schedule_relationships)
 
     {trip_ids, route_pattern_ids}
     |> case do
@@ -245,6 +266,25 @@ defmodule ApiWeb.PredictionController do
     end
   end
 
+  defp add_schedule_relationship_matchers(matchers, nil),
+    do: matchers
+
+  defp add_schedule_relationship_matchers(matchers, []),
+    do: matchers
+
+  defp add_schedule_relationship_matchers(matchers, schedule_relationships) do
+    for schedule_relationship <- schedule_relationships, matcher <- matchers do
+      schedule_relationship_atom =
+        schedule_relationship |> String.downcase() |> String.to_existing_atom()
+
+      if schedule_relationship_atom == :scheduled do
+        Map.put(matcher, :schedule_relationship, nil)
+      else
+        Map.put(matcher, :schedule_relationship, schedule_relationship_atom)
+      end
+    end
+  end
+
   def swagger_definitions do
     import PhoenixSwagger.JsonApi, except: [page: 1]
 

apps/api_web/lib/api_web/params.ex

@@ -201,6 +201,18 @@ defmodule ApiWeb.Params do
   def route_types(%{"route_type" => route_types}), do: integer_values(route_types)
   def route_types(_), do: []
 
+  @doc """
+  Parses a list of schedule relationships out of a parameter map
+  """
+  @spec schedule_relationships(%{String.t() => String.t()}) :: [String.t()] | nil
+  def schedule_relationships(%{"schedule_relationship" => schedule_relationships}),
+    do:
+      schedule_relationships
+      |> split_on_comma()
+      |> Enum.filter(&(&1 in ["SCHEDULED", "SKIPPED", "ADDED"]))
+
+  def schedule_relationships(_), do: nil
+
   @doc """
   Parse canonical filter param into boolean
   """

apps/api_web/lib/api_web/swagger_helpers.ex

@@ -285,6 +285,23 @@ defmodule ApiWeb.SwaggerHelpers do
     )
   end
 
+  def filter_param(path_object, :schedule_relationship, opts) do
+    Path.parameter(
+      path_object,
+      "filter[schedule_relationship]",
+      :query,
+      :string,
+      """
+      #{opts[:desc]}
+      Relationship between the prediction and the current GTFS static schedule.
+      When filter is not included, the default behavior is to return trips with all
+      schedule relationships.
+
+      """,
+      enum: ["SCHEDULED", "SKIPPED", "ADDED"]
+    )
+  end
+
   def page(resource) do
     resource
     |> JsonApi.page()

apps/api_web/lib/api_web/views/error_view.ex

@@ -47,6 +47,23 @@ defmodule ApiWeb.ErrorView do
     })
   end
 
+  def render("400.json" <> _, %{error: :only_schedule_relationship}) do
+    ErrorSerializer.format(%{
+      code: :bad_request,
+      status: "400",
+      detail: "filter[schedule_relationship] must be used in conjunction with another filter[]."
+    })
+  end
+
+  def render("400.json" <> _, %{error: :only_route_type_and_schedule_relationship}) do
+    ErrorSerializer.format(%{
+      code: :bad_request,
+      status: "400",
+      detail:
+        "filter[route_type], filter[schedule_relationship] must be used in conjunction with another filter[]."
+    })
+  end
+
   def render("400.json" <> _, %{error: :only_direction_id}) do
     ErrorSerializer.format(%{
       code: :bad_request,

apps/api_web/test/api_web/controllers/prediction_controller_test.exs

@@ -60,6 +60,20 @@ defmodule ApiWeb.PredictionControllerTest do
       assert ApiWeb.PredictionController.index_data(conn, %{"route_type" => "0,1"}) ==
                {:error, :only_route_type}
     end
+
+    test "returns an error if only the schedule_relationship filter is provided", %{conn: conn} do
+      assert ApiWeb.PredictionController.index_data(conn, %{"schedule_relationship" => "SKIPPED"}) ==
+               {:error, :only_schedule_relationship}
+    end
+
+    test "returns an error if only the schedule_relationship filter and route_type filter are provided",
+         %{conn: conn} do
+      assert ApiWeb.PredictionController.index_data(conn, %{
+               "schedule_relationship" => "SKIPPED",
+               "route_type" => "0,1"
+             }) ==
+               {:error, :only_route_type_and_schedule_relationship}
+    end
   end
 
   test "predictions can be paginated and are sorted by arrival_time", %{conn: base_conn} do
@@ -142,6 +156,39 @@ defmodule ApiWeb.PredictionControllerTest do
     end
   end
 
+  test "allows filtering by schedule_relationship", %{conn: conn} do
+    scheduled_prediction = %Prediction{
+      stop_id: "1",
+      route_id: "Red",
+      arrival_time: @latest_arrival,
+      schedule_relationship: nil
+    }
+
+    added_prediction = %Prediction{
+      stop_id: "1",
+      route_id: "Red",
+      schedule_relationship: :added
+    }
+
+    skipped_prediction = %Prediction{
+      stop_id: "1",
+      route_id: "Red",
+      schedule_relationship: :skipped
+    }
+
+    State.Prediction.new_state([added_prediction, skipped_prediction, scheduled_prediction])
+
+    for {params, expected} <- [
+          # show all scehdule_relationship values by default
+          {%{"route" => "Red"}, [added_prediction, skipped_prediction, scheduled_prediction]},
+          {%{"route" => "Red", "schedule_relationship" => "SKIPPED"}, [skipped_prediction]},
+          {%{"route" => "Red", "schedule_relationship" => "ADDED"}, [added_prediction]}
+        ] do
+      conn = get(conn, "/predictions", params)
+      assert conn.assigns.data == expected
+    end
+  end
+
   test "versions before 2021-01-09 allow an unused `date` filter", %{conn: conn} do
     conn = assign(conn, :api_version, "2021-01-09")
     resp = get(conn, "/predictions", stop: @stop.id, date: "2020-01-01")

apps/state/lib/state/prediction.ex

@@ -9,6 +9,18 @@ defmodule State.Prediction do
   import Parse.Time, only: [service_date: 1]
   import State.Route, only: [by_types: 1]
 
+  def filter_by_schedule_relationshp(predictions, nil), do: predictions
+  def filter_by_schedule_relationshp(predictions, []), do: predictions
+
+  def filter_by_schedule_relationshp(predictions, schedule_relationshps) do
+    route_ids =
+      schedule_relationshps
+      |> by_types()
+      |> MapSet.new(& &1.id)
+
+    Enum.filter(predictions, &(&1.route_id in route_ids))
+  end
+
   def filter_by_route_type(predictions, nil), do: predictions
   def filter_by_route_type(predictions, []), do: predictions