Add signs config fetching from live data

config/config.exs

@@ -38,7 +38,10 @@ config :concentrate,
     gtfs_realtime: [
       vehicle_positions: "https://cdn.mbta.com/realtime/VehiclePositions.pb",
       trip_updates: "https://cdn.mbta.com/realtime/TripUpdates.pb"
-    ]
+    ],
+  ],
+  signs_stops_config: [
+    url: "https://mbta-signs.s3.amazonaws.com/stops-config.json"
   ],
   alerts: [
     url: "https://cdn.mbta.com/realtime/Alerts.pb"
@@ -74,8 +77,8 @@ config :concentrate,
     Concentrate.GroupFilter.VehicleStopMatch,
     Concentrate.GroupFilter.SkippedStopOnAddedTrip,
     Concentrate.GroupFilter.TripDescriptorTimestamp,
-    Concentrate.GroupFilter.UncertaintyValue
-    # Concentrate.GroupFilter.SuppressStopTimeUpdate Enable once fully implemented
+    Concentrate.GroupFilter.UncertaintyValue,
+    Concentrate.GroupFilter.SuppressStopTimeUpdate
   ],
   source_reporters: [
     Concentrate.SourceReporter.Basic,

lib/concentrate.ex

@@ -85,6 +85,10 @@ defmodule Concentrate do
     [gtfs: [url: url]]
   end
 
+  defp decode_json_key_value({"signs_stops_config", %{"url" => url}}) do
+    [signs_stops_config: [url: url]]
+  end
+
   defp decode_json_key_value({"sinks", sinks_object}) do
     sinks = decode_sinks_object(sinks_object, %{})
 

lib/concentrate/filter/suppress/supervisor.ex

@@ -0,0 +1,42 @@
+defmodule Concentrate.Filter.Suppress.Supervisor do
+  @moduledoc """
+  Supervisor for the extra servers needed for suppressing predictions based on signs config.
+
+  * HTTP producer to fetch the signs config
+  * Consumer / map of suppressed stops
+  """
+  @one_day 86_400_000
+
+  require Logger
+
+  def start_link(config) do
+    if config[:url] do
+      Supervisor.start_link(
+        [
+          {
+            Concentrate.producer_for_url(config[:url]),
+            {
+              config[:url],
+              parser: Concentrate.Parser.SignsConfig,
+              fetch_after: 10_000,
+              content_warning_timeout: @one_day,
+              name: :stop_prediction_status_producer
+            }
+          },
+          {Concentrate.Filter.Suppress.StopPredictionStatus, subscribe_to: [:stop_prediction_status_producer]}
+        ],
+        strategy: :rest_for_one
+      )
+    else
+      :ignore
+    end
+  end
+
+  def child_spec(opts) do
+    %{
+      id: __MODULE__,
+      start: {__MODULE__, :start_link, [opts]},
+      type: :supervisor
+    }
+  end
+end

lib/concentrate/supervisor.ex

@@ -16,10 +16,11 @@ defmodule Concentrate.Supervisor do
   def children(config) do
     pool = pool()
     alerts = alerts(config[:alerts])
+    signs_stops_config = signs_stops_config(config[:signs_stops_config])
     gtfs = gtfs(config[:gtfs])
     pipeline = pipeline(config)
     health = health()
-    Enum.concat([pool, alerts, gtfs, pipeline, health])
+    Enum.concat([pool, alerts, signs_stops_config, gtfs, pipeline, health])
   end
 
   def pool do
@@ -34,6 +35,12 @@ defmodule Concentrate.Supervisor do
     ]
   end
 
+  def signs_stops_config(config) do
+    [
+      {Concentrate.Filter.Suppress.Supervisor, config}
+    ]
+  end
+
   def gtfs(config) do
     [
       {Concentrate.GTFS.Supervisor, config}