feat: unpredicted/unrealized predictions dashboard

* chore: bump phoenix_live_reload

* chore: Mix.Config is deprecated

* chore: bump ecto version for exists clause

assets/css/analyzer.scss

@@ -196,3 +196,8 @@
 .qs-datepicker-container {
   font-size: 1.5rem;
 }
+
+.departure_tables {
+  display: flex;
+  flex: 1;
+}

config/config.exs

@@ -1,9 +1,9 @@
 # This file is responsible for configuring your application
-# and its dependencies with the aid of the Mix.Config module.
+# and its dependencies with the aid of the Config module.
 #
 # This configuration file is loaded before any dependency and
 # is restricted to this project.
-use Mix.Config
+import Config
 
 # Configures the endpoint
 config :prediction_analyzer, PredictionAnalyzerWeb.Endpoint,
@@ -35,8 +35,6 @@ config :prediction_analyzer, :max_dwell_time_sec, 30 * 60
 config :prediction_analyzer, :prune_lookback_sec, 7 * 24 * 60 * 60
 config :prediction_analyzer, :analysis_lookback_min, 40
 
-config :prediction_analyzer, PredictionAnalyzer.Repo, adapter: Ecto.Adapters.Postgres
-
 config :prediction_analyzer, start_workers: true
 
 config :elixir, :time_zone_database, Tzdata.TimeZoneDatabase

config/dev.exs

@@ -1,4 +1,4 @@
-use Mix.Config
+import Config
 
 # For development, we disable any cache and enable
 # debugging and code reloading.
@@ -51,7 +51,7 @@ config :prediction_analyzer, PredictionAnalyzerWeb.Endpoint,
   ]
 
 # Do not include metadata nor timestamps in development logs
-config :logger, :console, format: "[$level] $message\n"
+config :logger, :console, format: "[$level] $message\n", level: :error
 
 # Set a higher stacktrace during development. Avoid configuring such
 # in production as building large stacktraces may be expensive.

config/prod.exs

@@ -1,4 +1,4 @@
-use Mix.Config
+import Config
 
 # For production, we often load configuration from external
 # sources, such as your system environment. For this reason,
@@ -24,7 +24,6 @@ config :logger,
   level: :debug
 
 config :prediction_analyzer, PredictionAnalyzer.Repo,
-  adapter: Ecto.Adapters.Postgres,
   pool_size: 15,
   ssl: true
 

config/runtime.exs

@@ -31,6 +31,8 @@ if config_env() == :prod do
     hostname: System.get_env("DATABASE_HOST"),
     port: port,
     pool_size: pool_size,
+    timeout: 60_000,
+    pool_timeout: 60_000,
     # password set by `configure` callback below
     configure: {PredictionAnalyzer.Repo, :before_connect, []}
 end

config/test.exs

@@ -1,4 +1,4 @@
-use Mix.Config
+import Config
 
 # We don't run a server during test. If one is required,
 # you can enable the server option below.

lib/prediction_analyzer/filters/stop_groups.ex

@@ -9,7 +9,7 @@ defmodule PredictionAnalyzer.Filters.StopGroups do
   @c_branch ~w(70211 70212 70213 70214 70215 70216 70217 70218 70219 70220 70223 70224 70225 70226 70227 70228 70229 70230 70231 70232 70233 70234 70235 70236 70237 70238)
   @d_branch ~w(70160 70161 70162 70163 70164 70165 70166 70167 70168 70169 70170 70171 70172 70173 70174 70175 70176 70177 70178 70179 70180 70181 70182 70183 70186 70187)
   @e_branch ~w(70239 70240 70241 70242 70243 70244 70245 70246 70247 70248 70249 70250 70251 70252 70253 70254 70255 70256 70257 70258 70260)
-  @terminals ~w(70001 70036 70038 70059 70060 70061 70093 70094 70105 70106 70107 70160 70161 70196 70197 70198 70199 70201 70202 70205 70206 70237 70238 70260 70261 70275 70276 70838 71199)
+  @terminals ~w(70001 70036 70038 70059 70061 70094 70105 70106 70160 70161 70196 70197 70198 70199 70201 70202 70205 70206 70238 70260 70261 70276 71199 70503)
 
   # Terminal departure platforms + away-from-terminal platforms up to 3 stops away. Note: 70001,
   # 70036, 70061, 70105, 70260, and 70261 include all platforms at the stop, since these terminals

lib/prediction_analyzer/missed_predictions.ex

@@ -0,0 +1,252 @@
+defmodule PredictionAnalyzer.MissedPredictions do
+  import Ecto.Query, only: [from: 2]
+  alias PredictionAnalyzer.VehicleEvents.VehicleEvent
+  alias PredictionAnalyzer.Predictions.Prediction
+  alias PredictionAnalyzer.Filters.StopGroups
+  alias PredictionAnalyzer.StopNameFetcher
+  import Utilities.Time, only: [format_unix: 1]
+
+  def unpredicted_departures_summary(date, env) do
+    {min_time, max_time} = service_times(date)
+
+    unpredicted_departures_query =
+      from(ve in VehicleEvent,
+        as: :vehicle_event,
+        where:
+          not is_nil(ve.trip_id) and
+            not is_nil(ve.departure_time) and
+            not ve.is_deleted and
+            ve.environment == ^env and
+            ve.departure_time >= ^min_time and
+            ve.departure_time <= ^max_time and
+            ve.stop_id in ^terminal_stops(),
+        group_by: ve.route_id,
+        select:
+          {ve.route_id, count(ve.id),
+           count(ve.id)
+           |> filter(
+             not exists(
+               from(p in Prediction,
+                 where:
+                   p.vehicle_event_id == parent_as(:vehicle_event).id and
+                     p.file_timestamp < parent_as(:vehicle_event).departure_time and
+                     not is_nil(p.departure_time)
+               )
+             )
+           ),
+           (count(ve.id)
+            |> filter(
+              not exists(
+                from(p in Prediction,
+                  where:
+                    p.vehicle_event_id == parent_as(:vehicle_event).id and
+                      p.file_timestamp < parent_as(:vehicle_event).departure_time and
+                      not is_nil(p.departure_time)
+                )
+              )
+            )) * 100.0 / count(ve.id)}
+      )
+
+    unpredicted_departures_query
+    |> PredictionAnalyzer.Repo.all()
+    |> Enum.sort_by(&Map.get(sort_map(), elem(&1, 0), 0))
+  end
+
+  def unpredicted_departures_for_route(date, env, route_id) do
+    {min_time, max_time} = service_times(date)
+
+    unpredicted_departures_query =
+      from(ve in VehicleEvent,
+        as: :vehicle_event,
+        where:
+          not is_nil(ve.trip_id) and
+            not is_nil(ve.departure_time) and
+            not ve.is_deleted and
+            ve.environment == ^env and
+            ve.departure_time >= ^min_time and
+            ve.departure_time <= ^max_time and
+            ve.stop_id in ^terminal_stops() and
+            ve.route_id == ^route_id,
+        group_by: ve.stop_id,
+        order_by: ve.stop_id,
+        select:
+          {ve.stop_id, count(ve.id),
+           count(ve.id)
+           |> filter(
+             not exists(
+               from(p in Prediction,
+                 where:
+                   p.vehicle_event_id == parent_as(:vehicle_event).id and
+                     p.file_timestamp < parent_as(:vehicle_event).departure_time and
+                     not is_nil(p.departure_time)
+               )
+             )
+           ),
+           (count(ve.id)
+            |> filter(
+              not exists(
+                from(p in Prediction,
+                  where:
+                    p.vehicle_event_id == parent_as(:vehicle_event).id and
+                      p.file_timestamp < parent_as(:vehicle_event).departure_time and
+                      not is_nil(p.departure_time)
+                )
+              )
+            )) * 100.0 / count(ve.id)}
+      )
+
+    unpredicted_departures_query
+    |> PredictionAnalyzer.Repo.all()
+    |> add_stop_names()
+  end
+
+  def unpredicted_departures_for_route_stop(date, env, route_id, stop_id) do
+    {min_time, max_time} = service_times(date)
+
+    unpredicted_departures_query =
+      from(ve in VehicleEvent,
+        as: :vehicle_event,
+        where:
+          not is_nil(ve.trip_id) and
+            not is_nil(ve.departure_time) and
+            not ve.is_deleted and
+            ve.environment == ^env and
+            ve.departure_time >= ^min_time and
+            ve.departure_time <= ^max_time and
+            ve.route_id == ^route_id and
+            ve.stop_id == ^stop_id and
+            not exists(
+              from(p in Prediction,
+                where:
+                  p.vehicle_event_id == parent_as(:vehicle_event).id and
+                    p.file_timestamp < parent_as(:vehicle_event).departure_time and
+                    not is_nil(p.departure_time)
+              )
+            ),
+        order_by: [asc: ve.departure_time],
+        select: {ve.vehicle_id, ve.trip_id, ve.departure_time}
+      )
+
+    unpredicted_departures_query
+    |> PredictionAnalyzer.Repo.all()
+    |> Enum.map(fn row ->
+      {
+        elem(row, 0),
+        elem(row, 1),
+        elem(row, 2) |> format_unix()
+      }
+    end)
+  end
+
+  def missed_departures_summary(date, env) do
+    {min_time, max_time} = service_times(date)
+
+    missed_departures_query =
+      from(p in Prediction,
+        where:
+          p.environment == ^env and
+            p.file_timestamp >= ^min_time and
+            p.file_timestamp <= ^max_time and
+            not is_nil(p.departure_time) and
+            p.stop_id in ^terminal_stops(),
+        group_by: p.route_id,
+        select:
+          {p.route_id, count(p.trip_id, :distinct),
+           count(p.trip_id, :distinct) |> filter(is_nil(p.vehicle_event_id)),
+           (count(p.trip_id, :distinct) |> filter(is_nil(p.vehicle_event_id))) * 100.0 /
+             count(p.trip_id, :distinct)}
+      )
+
+    missed_departures_query
+    |> PredictionAnalyzer.Repo.all()
+    |> Enum.sort_by(&Map.get(sort_map(), elem(&1, 0), 0))
+  end
+
+  def missed_departures_for_route(date, env, route_id) do
+    {min_time, max_time} = service_times(date)
+
+    missed_departures_query =
+      from(p in Prediction,
+        where:
+          p.environment == ^env and
+            p.file_timestamp >= ^min_time and
+            p.file_timestamp <= ^max_time and
+            not is_nil(p.departure_time) and
+            p.route_id == ^route_id and
+            p.stop_id in ^terminal_stops(),
+        group_by: p.stop_id,
+        order_by: [desc: count(p.trip_id, :distinct)],
+        select:
+          {p.stop_id, count(p.trip_id, :distinct),
+           count(p.trip_id, :distinct) |> filter(is_nil(p.vehicle_event_id)),
+           (count(p.trip_id, :distinct) |> filter(is_nil(p.vehicle_event_id))) * 100.0 /
+             count(p.trip_id, :distinct)}
+      )
+
+    PredictionAnalyzer.Repo.all(missed_departures_query) |> add_stop_names()
+  end
+
+  def missed_departures_for_route_stop(date, env, route_id, stop_id) do
+    {min_time, max_time} = service_times(date)
+
+    missed_departures_query =
+      from(p in Prediction,
+        where:
+          p.environment == ^env and
+            p.file_timestamp >= ^min_time and
+            p.file_timestamp <= ^max_time and
+            p.route_id == ^route_id and
+            not is_nil(p.departure_time) and
+            is_nil(p.vehicle_event_id) and
+            p.stop_id == ^stop_id,
+        group_by: [p.vehicle_id, p.trip_id],
+        order_by: [p.vehicle_id, min(p.departure_time)],
+        select:
+          {p.vehicle_id, p.trip_id, min(p.departure_time), max(p.departure_time),
+           min(p.file_timestamp), max(p.file_timestamp), count(p.id)}
+      )
+
+    missed_departures_query
+    |> PredictionAnalyzer.Repo.all()
+    |> Enum.map(fn row ->
+      {
+        elem(row, 0),
+        elem(row, 1),
+        elem(row, 2) |> format_unix(),
+        elem(row, 3) |> format_unix(),
+        elem(row, 4) |> format_unix(),
+        elem(row, 5) |> format_unix(),
+        elem(row, 6)
+      }
+    end)
+  end
+
+  defp service_times(date) do
+    start_time = Time.new!(4, 0, 0)
+    {:ok, start_date_time} = DateTime.new(date, start_time, "America/New_York")
+
+    end_time = Time.new!(2, 0, 0)
+    tomorrow = Date.add(date, 1)
+    {:ok, end_date_time} = DateTime.new(tomorrow, end_time, "America/New_York")
+
+    {start_date_time |> DateTime.to_unix(), end_date_time |> DateTime.to_unix()}
+  end
+
+  defp sort_map() do
+    PredictionAnalyzer.Utilities.routes_for_mode(:subway) |> Enum.with_index() |> Enum.into(%{})
+  end
+
+  defp terminal_stops() do
+    StopGroups.expand_groups(["_terminal"])
+  end
+
+  defp add_stop_names(data) do
+    stop_dict = StopNameFetcher.get_stop_descriptions(:subway)
+
+    Enum.map(data, fn row ->
+      stop_id = elem(row, 0)
+      stop_name = Map.get(stop_dict, stop_id, stop_id)
+      Tuple.insert_at(row, 1, stop_name)
+    end)
+  end
+end

lib/prediction_analyzer/repo.ex

@@ -1,5 +1,5 @@
 defmodule PredictionAnalyzer.Repo do
-  use Ecto.Repo, otp_app: :prediction_analyzer
+  use Ecto.Repo, otp_app: :prediction_analyzer, adapter: Ecto.Adapters.Postgres
   require Logger
 
   @doc """

lib/prediction_analyzer/vehicle_positions/comparator.ex

@@ -92,9 +92,10 @@ defmodule PredictionAnalyzer.VehiclePositions.Comparator do
         ve.environment == ^vehicle.environment and ve.vehicle_id == ^vehicle.id and
           ve.stop_id == ^vehicle.stop_id and is_nil(ve.departure_time) and
           ve.arrival_time > ^(System.system_time(:second) - max_dwell_time_sec),
-      update: [set: [departure_time: ^vehicle.timestamp]]
+      update: [set: [departure_time: ^vehicle.timestamp]],
+      select: ve
     )
-    |> Repo.update_all([], returning: true)
+    |> Repo.update_all([])
     |> case do
       {0, _} ->
         Logger.warn("Tried to update departure time, but no arrival for #{vehicle.label}")

lib/prediction_analyzer_web.ex

@@ -41,6 +41,8 @@ defmodule PredictionAnalyzerWeb do
       alias PredictionAnalyzerWeb.Router.Helpers, as: Routes
       import PredictionAnalyzerWeb.ErrorHelpers
       import PredictionAnalyzerWeb.Gettext
+      import PredictionAnalyzerWeb.ViewHelpers
+      import Utilities.Time
     end
   end