feat: Support analysis over multiple service days (#9)

* feat: Support analysis over multiple service days

* Update notebooks to support analysis over multiple service days

* fix: Use aliased module name

* Add more thorough documentation of output tables

Author: jzimbel-mbta

lib/transit_data/glides_report/departure.ex

@@ -6,27 +6,26 @@ defmodule TransitData.GlidesReport.Departure do
 
   alias TransitData.GlidesReport.Spec.Common
   alias TransitData.GlidesReport.Terminal
-  alias TransitData.GlidesReport.Util
 
   @type t :: %__MODULE__{
           trip: Common.trip_id(),
           terminal: Terminal.id(),
           timestamp: Common.timestamp(),
-          # Hour part of the timestamp (in Eastern TZ)
-          hour: 0..23,
-          # Minute part of the timestamp
-          minute: 0..59
+          # `timestamp` as a DateTime in Eastern time, truncated to minutes
+          local_dt: DateTime.t()
         }
 
-  @type minute :: 0..59
-
-  @enforce_keys [:trip, :terminal, :timestamp, :hour, :minute]
+  @enforce_keys [:trip, :terminal, :timestamp, :local_dt]
   defstruct @enforce_keys
 
   @spec new(Common.trip_id(), Terminal.id(), Common.timestamp()) :: t()
   def new(trip, {:terminal, _} = terminal, timestamp) do
-    hour = Util.unix_timestamp_to_local_hour(timestamp)
-    minute = Util.unix_timestamp_to_local_minute(timestamp)
-    %__MODULE__{trip: trip, terminal: terminal, timestamp: timestamp, hour: hour, minute: minute}
+    local_dt =
+      timestamp
+      |> DateTime.from_unix!()
+      |> DateTime.shift_zone!("America/New_York")
+      |> then(&%{&1 | second: 0, microsecond: {0, 0}})
+
+    %__MODULE__{trip: trip, terminal: terminal, timestamp: timestamp, local_dt: local_dt}
   end
 end

lib/transit_data/glides_report/loader.ex

@@ -10,17 +10,12 @@ defmodule TransitData.GlidesReport.Loader do
 
   @doc """
   Loads data into ETS tables, and returns counts of files found locally vs downloaded.
-
-  `start_dt` and `end_dt` are NaiveDateTimes assumed to be in UTC.
   """
-  @spec load_data(
-          NaiveDateTime.t(),
-          NaiveDateTime.t(),
-          String.t(),
-          pos_integer,
-          pos_integer | :all
-        ) :: %{local: non_neg_integer, downloaded: non_neg_integer}
-  def load_data(start_dt, end_dt, env_suffix, sample_rate, sample_count) do
+  @spec load_data(Date.t(), Date.t(), String.t(), pos_integer, pos_integer | :all) :: %{
+          local: non_neg_integer,
+          downloaded: non_neg_integer
+        }
+  def load_data(start_date, end_date, env_suffix, sample_rate, sample_count) do
     dir = local_dir(env_suffix)
 
     IO.puts(
@@ -35,17 +30,24 @@ defmodule TransitData.GlidesReport.Loader do
 
     s3_bucket = "mbta-gtfs-s3#{env_suffix}"
 
-    start_dt_utc = DateTime.from_naive!(start_dt, "Etc/UTC")
-    end_dt_utc = DateTime.from_naive!(end_dt, "Etc/UTC")
+    start_dt =
+      DateTime.new!(start_date, ~T[04:00:00], "America/New_York")
+      |> DateTime.shift_zone!("Etc/UTC")
 
-    total_minutes = DateTime.diff(end_dt_utc, start_dt_utc, :minute)
-    total_increments = div(total_minutes, sample_rate)
+    end_dt =
+      end_date
+      # The service day ends on the following calendar date.
+      |> Date.shift(day: 1)
+      |> DateTime.new!(~T[03:59:59], "America/New_York")
+      |> DateTime.shift_zone!("Etc/UTC")
+
+    total_minutes = DateTime.diff(end_dt, start_dt, :minute)
 
     # Prefixes used to list S3 objects timestamped within the same minute.
     minute_prefixes =
-      Enum.map(0..total_increments, fn increment ->
-        start_dt_utc
-        |> DateTime.add(increment * sample_rate, :minute)
+      Enum.map(0..total_minutes//sample_rate, fn increment ->
+        start_dt
+        |> DateTime.add(increment, :minute)
         |> Calendar.strftime("%Y/%m/%d/%Y-%m-%dT%H:%M")
       end)
 
@@ -68,27 +70,19 @@ defmodule TransitData.GlidesReport.Loader do
   end
 
   # Loads data into a table.
-  # Returns the number of files that were found locally,
-  # the number that were newly downloaded,
-  # and a list of minutes for which not enough data could be found.
+  # Returns the number of files that were found locally and the number
+  # that were newly downloaded:
+  # %{local: integer, downloaded: integer}
   defp populate_table(table_name, path_prefixes, s3_bucket, sample_count) do
     IO.puts("Loading #{table_name}...")
 
+    prefix_count = length(path_prefixes)
+
     {total, insufficients} =
       path_prefixes
-      |> Stream.with_index(fn prefix, i ->
-        IO.write([
-          IO.ANSI.clear_line(),
-          "\r",
-          moons_of_progress()[rem(i, 8)],
-          " Loading data for ",
-          prefix_to_local_minute(prefix)
-        ])
-
-        prefix
-      end)
+      |> Stream.with_index(&update_progress(&1, &2, prefix_count))
       |> Task.async_stream(
-        &load_minute(&1, s3_bucket, table_name, sample_count),
+        fn prefix -> load_minute(prefix, s3_bucket, table_name, sample_count) end,
         ordered: false,
         timeout: 60_000
       )
@@ -101,7 +95,7 @@ defmodule TransitData.GlidesReport.Loader do
 
         insufficients =
           if is_integer(sample_count) and counts.local + counts.downloaded < sample_count,
-            do: [prefix_to_local_minute(counts.prefix) | insufficients],
+            do: [prefix_to_local_dt(counts.prefix) | insufficients],
             else: insufficients
 
         {total, insufficients}
@@ -110,20 +104,8 @@ defmodule TransitData.GlidesReport.Loader do
     IO.puts("#{IO.ANSI.clear_line()}\r🌝 Done")
 
     unless Enum.empty?(insufficients) do
-      time_ranges =
-        insufficients
-        |> Enum.sort()
-        |> Enum.split_while(&(&1 < "04"))
-        |> then(fn {after_midnight_service_day, service_day} ->
-          service_day ++ after_midnight_service_day
-        end)
-        |> Stream.map(&Time.from_iso8601!(&1 <> ":00"))
-        # Chunk the individual times into ranges of consecutive times for better human readability.
-        |> Stream.chunk_while(nil, &chunk_time_ranges/2, &{:cont, hh_mm_range(&1), nil})
-        |> Stream.reject(&is_nil/1)
-        |> Enum.join(", ")
-
-      IO.puts("#{table_name}: Insufficient data available for minute(s): #{time_ranges}")
+      time_ranges = datetimes_to_time_ranges(insufficients)
+      IO.puts("#{table_name}: Insufficient data available for minute(s):\n#{time_ranges}")
     end
 
     IO.puts("")
@@ -139,11 +121,60 @@ defmodule TransitData.GlidesReport.Loader do
     """x
   end
 
+  defp update_progress(prefix, i, total) do
+    pct =
+      (100 * i / total)
+      |> trunc()
+      |> Integer.to_string()
+      |> String.pad_leading(3)
+
+    IO.write([
+      IO.ANSI.clear_line(),
+      "\r",
+      moons_of_progress()[rem(i, 8)],
+      " Loading data for ",
+      Calendar.strftime(prefix_to_local_dt(prefix), "%x %H:%M"),
+      "  ",
+      pct,
+      "%"
+    ])
+
+    prefix
+  end
+
   # 🌝
   defp moons_of_progress do
     %{0 => "🌕", 1 => "🌖", 2 => "🌗", 3 => "🌘", 4 => "🌑", 5 => "🌒", 6 => "🌓", 7 => "🌔"}
   end
 
+  @doc ~S'''
+  Returns a human-readable string describing a list of minute-granularity
+  local-timezone DateTimes as comma-separated time ranges.
+
+      iex> [~U[2025-01-01T18:00:00Z], ~U[2025-01-01T18:01:00Z], ~U[2025-01-02T08:03:00Z], ~U[2025-01-02T12:00:00Z]]
+      ...> |> Enum.map(&DateTime.shift_zone!(&1, "America/New_York"))
+      ...> |> datetimes_to_time_ranges()
+      """
+      • 2025-01-01: 13:00-13:01, 03:03
+      • 2025-01-02: 07:00\
+      """
+  '''
+  def datetimes_to_time_ranges(datetimes) do
+    datetimes
+    |> Enum.sort(DateTime)
+    |> Stream.chunk_by(&service_day/1)
+    |> Stream.map(fn dts ->
+      time_ranges =
+        dts
+        |> Stream.map(fn %DateTime{time_zone: "America/New_York"} = dt -> DateTime.to_time(dt) end)
+        |> Stream.chunk_while(nil, &chunk_time_ranges/2, &{:cont, hh_mm_range(&1), nil})
+        |> Enum.join(", ")
+
+      "• #{service_day(hd(dts))}: #{time_ranges}"
+    end)
+    |> Enum.join("\n")
+  end
+
   defp chunk_time_ranges(time, nil) do
     # This is the first time in the list.
     {:cont, {time, time}}
@@ -182,11 +213,10 @@ defmodule TransitData.GlidesReport.Loader do
 
   defp hh_mm(time), do: Calendar.strftime(time, "%H:%M")
 
-  defp prefix_to_local_minute(prefix) do
+  defp prefix_to_local_dt(prefix) do
     prefix
     |> prefix_to_dt()
     |> DateTime.shift_zone!("America/New_York")
-    |> Calendar.strftime("%H:%M")
   end
 
   defp prefix_to_dt(prefix) do
@@ -199,6 +229,23 @@ defmodule TransitData.GlidesReport.Loader do
     dt
   end
 
+  def service_day(%{time_zone: "America/New_York"} = dt) do
+    # Service day starts at 4am.
+    # Times before that on a calendar date are part of the previous
+    # calendar date's service day.
+    if dt.hour >= 4 do
+      DateTime.to_date(dt)
+    else
+      dt |> DateTime.to_date() |> Date.add(-1)
+    end
+  end
+
+  def service_day(%DateTime{} = dt) do
+    dt
+    |> DateTime.shift_zone!("America/New_York")
+    |> service_day()
+  end
+
   # Loads data for a specific minute of the service day, either by reading existing local files,
   # by downloading and reading new files from S3, or a mix of both.
   #

lib/transit_data/glides_report/settings/load.ex

@@ -5,32 +5,20 @@ defmodule TransitData.GlidesReport.Settings.Load do
 
   @type t :: %__MODULE__{
           env_suffix: String.t(),
-          start_dt: DateTime.t(),
-          end_dt: DateTime.t(),
-          sample_rate: integer,
-          sample_count: integer | :all
+          start_date: Date.t(),
+          end_date: Date.t(),
+          sample_rate: pos_integer,
+          sample_count: pos_integer | :all
         }
 
-  defstruct [
-    :env_suffix,
-    :start_dt,
-    :end_dt,
-    :sample_rate,
-    :sample_count
-  ]
+  defstruct [:env_suffix, :start_date, :end_date, :sample_rate, :sample_count]
 
-  @spec new(String.t(), Date.t(), integer, integer | nil) :: t()
-  def new(env, date, sample_rate, samples_per_minute) do
-    {start_dt, end_dt} = date_to_start_end_dt(date)
-    new(env, start_dt, end_dt, sample_rate, samples_per_minute)
-  end
-
-  @spec new(String.t(), DateTime.t(), DateTime.t(), integer, integer | nil) :: t()
-  def new(env, start_dt, end_dt, sample_rate, samples_per_minute) do
+  @spec new(String.t(), Date.t(), Date.t(), integer, integer | nil) :: t()
+  def new(env, start_date, end_date, sample_rate, samples_per_minute) do
     %__MODULE__{
       env_suffix: env,
-      start_dt: start_dt,
-      end_dt: end_dt,
+      start_date: start_date,
+      end_date: end_date,
       sample_rate: sample_rate |> trunc(),
       sample_count:
         case samples_per_minute do
@@ -39,20 +27,4 @@ defmodule TransitData.GlidesReport.Settings.Load do
         end
     }
   end
-
-  defp date_to_start_end_dt(date) do
-    # We assume they want a full service day on the given date.
-    start_dt =
-      date
-      |> DateTime.new!(~T[04:00:00], "America/New_York")
-      |> DateTime.shift_zone!("Etc/UTC")
-
-    end_dt =
-      date
-      |> Date.add(1)
-      |> DateTime.new!(~T[03:59:59], "America/New_York")
-      |> DateTime.shift_zone!("Etc/UTC")
-
-    {start_dt, end_dt}
-  end
 end

lib/transit_data/glides_report/util.ex

@@ -100,22 +100,6 @@ defmodule TransitData.GlidesReport.Util do
     "#{p}%"
   end
 
-  @spec unix_timestamp_to_local_hour(integer) :: 0..23
-  def unix_timestamp_to_local_hour(timestamp) do
-    unix_timestamp_to_local_datetime(timestamp).hour
-  end
-
-  @spec unix_timestamp_to_local_hour(integer) :: 0..59
-  def unix_timestamp_to_local_minute(timestamp) do
-    unix_timestamp_to_local_datetime(timestamp).minute
-  end
-
-  defp unix_timestamp_to_local_datetime(timestamp) do
-    timestamp
-    |> DateTime.from_unix!()
-    |> DateTime.shift_zone!("America/New_York")
-  end
-
   @doc """
   Returns /absolute/path/to/transit_data_reports/dataset.
   """
@@ -154,8 +138,8 @@ defmodule TransitData.GlidesReport.Util do
   def build_csv_name(table_name, loader_settings, filter_settings) do
     %{
       env_suffix: env_suffix,
-      start_dt: start_dt,
-      end_dt: end_dt,
+      start_date: start_date,
+      end_date: end_date,
       sample_rate: sample_rate,
       sample_count: sample_count
     } = loader_settings
@@ -168,12 +152,9 @@ defmodule TransitData.GlidesReport.Util do
     env = if env_suffix == "", do: "prod", else: String.slice(env_suffix, 1..-1//1)
 
     dt_range =
-      [start_dt, end_dt]
-      |> Enum.map(&DateTime.shift_zone!(&1, "America/New_York"))
-      |> Enum.map_join(
-        "-",
-        &(&1 |> DateTime.shift_zone!("America/New_York") |> Calendar.strftime("%xT%H:%M"))
-      )
+      if Date.compare(start_date, end_date) == :eq,
+        do: "#{start_date}",
+        else: "#{start_date} to #{end_date}"
 
     terminals_filter =
       Enum.find_value(Terminal.labeled_terminal_groups(), fn {_id, label, terminals_in_group} ->

mix.exs

@@ -38,7 +38,7 @@ defmodule TransitData.MixProject do
       {:ex_aws_s3, "~> 2.5"},
       {:hackney, "~> 1.20"},
       {:jaxon, "~> 2.0"},
-      {:kino, "~> 0.12.0"},
+      {:kino, "~> 0.14.2"},
       {:stream_gzip, "~> 0.4.2"},
       {:sweet_xml, "~> 0.7.4"},
       # TEST-ENV DEPS