Skip to content

Service Ridership Dashboard Time Series

This page documents time series functions in the Service Ridership Dashboard module.

ingestor.chalicelib.service_ridership_dashboard.time_series

get_weekly_median_time_series(entries, entry_value_getter, start_date, max_end_date)

Compute a weekly median time series from daily entries.

Parameters:

Name Type Description Default
entries dict[date, Entry]

A dictionary mapping dates to entry values.

required
entry_value_getter Callable[[Entry], float]

A callable that extracts a float value from an entry.

required
start_date date

The start date of the time series.

required
max_end_date date

The maximum end date of the time series.

required

Returns:

Type Description
WeeklyMedianTimeSeries

A dictionary mapping date strings (yyyy-mm-dd) to weekly median values.

Source code in ingestor/chalicelib/service_ridership_dashboard/time_series.py
def get_weekly_median_time_series(
    entries: dict[date, Entry],
    entry_value_getter: Callable[[Entry], float],
    start_date: date,
    max_end_date: date,
) -> WeeklyMedianTimeSeries:
    """Compute a weekly median time series from daily entries.

    Args:
        entries: A dictionary mapping dates to entry values.
        entry_value_getter: A callable that extracts a float value from an entry.
        start_date: The start date of the time series.
        max_end_date: The maximum end date of the time series.

    Returns:
        A dictionary mapping date strings (yyyy-mm-dd) to weekly median values.
    """
    weekly_buckets = _bucket_by_week(entries)
    weekly_medians: dict[str, float] = {}
    for week_start, week_entries in _iterate_mondays(weekly_buckets, start_date, max_end_date):
        week_values = [entry_value_getter(entry) for entry in week_entries]
        week_values.sort()
        weekly_medians[date_to_string(week_start)] = week_values[len(week_values) // 2]
    return weekly_medians

merge_weekly_median_time_series(many_series)

Merge multiple weekly median time series by summing values for each week.

Parameters:

Name Type Description Default
many_series list[WeeklyMedianTimeSeries]

A list of WeeklyMedianTimeSeries dictionaries to merge.

required

Returns:

Type Description
WeeklyMedianTimeSeries

A single WeeklyMedianTimeSeries with summed values for each week.

Source code in ingestor/chalicelib/service_ridership_dashboard/time_series.py
def merge_weekly_median_time_series(many_series: list[WeeklyMedianTimeSeries]) -> WeeklyMedianTimeSeries:
    """Merge multiple weekly median time series by summing values for each week.

    Args:
        many_series: A list of WeeklyMedianTimeSeries dictionaries to merge.

    Returns:
        A single WeeklyMedianTimeSeries with summed values for each week.
    """
    merged_series: dict[str, float] = {}
    for series in many_series:
        for week_start, value in series.items():
            merged_series.setdefault(week_start, 0)
            merged_series[week_start] += value
    return merged_series

get_weekly_median_time_series_entry_for_date(series, date)

Look up the weekly median value for the week containing a given date.

Parameters:

Name Type Description Default
series WeeklyMedianTimeSeries

A WeeklyMedianTimeSeries dictionary.

required
date date

The date to look up.

required

Returns:

Type Description
Optional[float]

The median value for that week, or None if not found.

Source code in ingestor/chalicelib/service_ridership_dashboard/time_series.py
def get_weekly_median_time_series_entry_for_date(series: WeeklyMedianTimeSeries, date: date) -> Optional[float]:
    """Look up the weekly median value for the week containing a given date.

    Args:
        series: A WeeklyMedianTimeSeries dictionary.
        date: The date to look up.

    Returns:
        The median value for that week, or None if not found.
    """
    monday = _get_monday_of_week_containing_date(date)
    return series.get(date_to_string(monday))

get_latest_weekly_median_time_series_entry(series)

Get the value of the most recent week in a weekly median time series.

Parameters:

Name Type Description Default
series WeeklyMedianTimeSeries

A WeeklyMedianTimeSeries dictionary.

required

Returns:

Type Description
Optional[float]

The value for the latest week, or None if not found.

Source code in ingestor/chalicelib/service_ridership_dashboard/time_series.py
def get_latest_weekly_median_time_series_entry(series: WeeklyMedianTimeSeries) -> Optional[float]:
    """Get the value of the most recent week in a weekly median time series.

    Args:
        series: A WeeklyMedianTimeSeries dictionary.

    Returns:
        The value for the latest week, or None if not found.
    """
    latest_date = max(series.keys())
    return series.get(latest_date)

get_earliest_weekly_median_time_series_entry(series)

Get the value of the earliest week in a weekly median time series.

Parameters:

Name Type Description Default
series WeeklyMedianTimeSeries

A WeeklyMedianTimeSeries dictionary.

required

Returns:

Type Description
Optional[float]

The value for the earliest week, or None if not found.

Source code in ingestor/chalicelib/service_ridership_dashboard/time_series.py
def get_earliest_weekly_median_time_series_entry(series: WeeklyMedianTimeSeries) -> Optional[float]:
    """Get the value of the earliest week in a weekly median time series.

    Args:
        series: A WeeklyMedianTimeSeries dictionary.

    Returns:
        The value for the earliest week, or None if not found.
    """
    earliest_date = min(series.keys())
    return series.get(earliest_date)