Skip to content

Service Ridership Dashboard Service Levels

This page documents service level functions in the Service Ridership Dashboard module.

ingestor.chalicelib.service_ridership_dashboard.service_levels

get_service_level_entries_by_line_id(routes_by_line, start_date, end_date)

Query and aggregate service level data for all lines, organized by line ID and date.

Parameters:

Name Type Description Default
routes_by_line RoutesByLine

A dictionary mapping Line objects to their associated routes.

 required
start_date date

The start date of the query range.

 required
end_date date

The end date of the query range.

 required

Returns:

Type Description
ServiceLevelsByLineId

A dictionary mapping line IDs to dictionaries of date-to-ServiceLevelsEntry mappings.
Source code in ingestor/chalicelib/service_ridership_dashboard/service_levels.py 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def get_service_level_entries_by_line_id(
    routes_by_line: RoutesByLine,
    start_date: date,
    end_date: date,
) -> ServiceLevelsByLineId:
    """Query and aggregate service level data for all lines, organized by line ID and date.

    Args:
        routes_by_line: A dictionary mapping Line objects to their associated routes.
        start_date: The start date of the query range.
        end_date: The end date of the query range.

    Returns:
        A dictionary mapping line IDs to dictionaries of date-to-ServiceLevelsEntry mappings.
    """
    entries: dict[str, list[ServiceLevelsEntry]] = {}
    for line, routes in (progress := tqdm(routes_by_line.items())):
        entries.setdefault(line.line_id, [])
        progress.set_description(f"Loading service levels for {line.line_id}")
        results_by_date_str: dict[str, list[ScheduledServiceRow]] = bucket_by(
            [
                row
                for route in routes
                for row in query_scheduled_service(
                    start_date=start_date,
                    end_date=end_date,
                    route_id=route.route_id,
                )
            ],
            lambda row: row["date"],
        )
        for today in date_range(start_date, end_date):
            today_str = date_to_string(today)
            all_service_levels_today = results_by_date_str.get(today_str, [])
            entry = ServiceLevelsEntry(
                date=today,
                line_id=line.line_id,
                line_short_name=line.line_short_name,
                line_long_name=line.line_long_name,
                route_ids=[route.route_id for route in routes],
                service_levels=_get_trip_count_by_hour_totals_for_day(all_service_levels_today),
                has_service_exceptions=_get_has_service_exception(all_service_levels_today),
            )
            entries[line.line_id].append(entry)
    return {line_id: index_by(entries, lambda e: e.date) for line_id, entries in entries.items()}