Skip to content

Service Ridership Dashboard GTFS

This page documents gtfs functions in the Service Ridership Dashboard module.

ingestor.chalicelib.service_ridership_dashboard.gtfs

get_routes_by_line(include_only_line_ids)

Fetch routes from the latest GTFS feed and group them by their parent line.

Parameters:

Name Type Description Default
include_only_line_ids Optional[list[str]]

If provided, only include routes belonging to these line IDs.

 required

Returns:

Type Description
dict[Line, Route]

A dictionary mapping Line objects to their associated Route objects.
Source code in ingestor/chalicelib/service_ridership_dashboard/gtfs.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def get_routes_by_line(include_only_line_ids: Optional[list[str]]) -> dict[Line, Route]:
    """Fetch routes from the latest GTFS feed and group them by their parent line.

    Args:
        include_only_line_ids: If provided, only include routes belonging to these line IDs.

    Returns:
        A dictionary mapping Line objects to their associated Route objects.
    """
    s3 = boto3.resource("s3")
    archive = MbtaGtfsArchive(
        local_archive_path=TemporaryDirectory().name,
        s3_bucket=s3.Bucket("tm-gtfs"),
    )
    feed = archive.get_latest_feed()
    feed.use_compact_only()
    feed.download_or_build()
    session = feed.create_sqlite_session(compact=True)
    lines_by_id = index_by(
        session.query(Line).all(),
        lambda line: line.line_id,
    )
    all_routes_with_line_ids = [
        route
        for route in session.query(Route).all()
        if route.line_id
        and route.line_id not in IGNORE_LINE_IDS
        and (not include_only_line_ids or route.line_id in include_only_line_ids)
    ]
    return bucket_by(all_routes_with_line_ids, lambda route: lines_by_id[route.line_id])