Stops

Stops

Stops(lf: Optional[LazyFrame] = None, gdf: Optional[GeoDataFrame] = None, stop_ids: Optional[List[str]] = None, mean_lon: Optional[float] = None, mean_lat: Optional[float] = None)

A class to manage GTFS stops using Polars LazyFrames and GeoPandas GeoDataFrames.

Provides functionality to: - Read and optionally filter GTFS stops from one or more directories - Filter stops by area of interest (AOI) - Group stops spatially and assign parent_station values

Attributes:

Name	Type	Description
lf	LazyFrame	LazyFrame containing GTFS stops.
gdf	GeoDataFrame	GeoDataFrame of stop_id and geometry.
stop_ids	list[str]	List of stop IDs currently loaded.
paths	List[Path]	List of GTFS paths (directories).

Wraps already-loaded stops frames, or leaves them empty for load to fill in.

Parameters:

Name	Type	Description	Default
lf	Optional[LazyFrame]	Optional pre-loaded stops.txt LazyFrame.	None
gdf	Optional[GeoDataFrame]	Optional pre-built GeoDataFrame of stop_id + geometry.	None
stop_ids	Optional[List[str]]	Optional pre-computed list of in-scope stop_ids.	None
mean_lon	Optional[float]	Optional mean stop longitude (used for e.g. holiday lookups elsewhere that need an approximate feed location).	None
mean_lat	Optional[float]	Optional mean stop latitude.	None

Source code in pyGTFSHandler/models/stops.py

def __init__(
    self,
    lf: Optional[pl.LazyFrame] = None,
    gdf: Optional[gpd.GeoDataFrame] = None,
    stop_ids: Optional[List[str]] = None,
    mean_lon: Optional[float] = None,
    mean_lat: Optional[float] = None,
) -> None:
    """Wraps already-loaded stops frames, or leaves them empty for `load` to fill in.

    Args:
        lf: Optional pre-loaded `stops.txt` LazyFrame.
        gdf: Optional pre-built GeoDataFrame of `stop_id` + geometry.
        stop_ids: Optional pre-computed list of in-scope `stop_id`s.
        mean_lon: Optional mean stop longitude (used for e.g. holiday
            lookups elsewhere that need an approximate feed location).
        mean_lat: Optional mean stop latitude.
    """
    self.lf = lf
    self.gdf = gdf 
    self.stop_ids = stop_ids
    if (mean_lon is None) or (mean_lat is None):
        if lf is not None: 
            mean_coords = lf.select(
                [
                    pl.col("stop_lon").mean().alias("mean_lon"),
                    pl.col("stop_lat").mean().alias("mean_lat"),
                ]
            ).collect()

            mean_lon = mean_coords["mean_lon"][0]
            mean_lat = mean_coords["mean_lat"][0]

    self.mean_lon = mean_lon 
    self.mean_lat = mean_lat 

load

load(path: Union[str, Path, List[Union[str, Path]]], aoi: Union[GeoDataFrame, GeoSeries, None] = None, stop_group_distance: float = 0, stop_ids: Union[List[str], DataFrame | LazyFrame] = None, check_files: bool = False, min_file_id: int = 0, collision_registry=None)

Initialize Stops instance and load GTFS stops from one or more files.

Parameters:

Name	Type	Description	Default
path	str | Path | list	One or more paths to GTFS directories.	required
aoi	GeoDataFrame | GeoSeries	Area of interest for spatial filtering.	None
stop_ids	list[str]	List of stop IDs to include.	None

Source code in pyGTFSHandler/models/stops.py

def load(
    self,
    path: Union[str, Path, List[Union[str, Path]]],
    aoi: Union[gpd.GeoDataFrame, gpd.GeoSeries, None] = None,
    stop_group_distance: float = 0,
    stop_ids: Union[List[str], pl.DataFrame | pl.LazyFrame] = None,
    check_files:bool=False,
    min_file_id:int=0,
    collision_registry=None,
):
    """
    Initialize Stops instance and load GTFS stops from one or more files.

    Args:
        path (str | Path | list): One or more paths to GTFS directories.
        aoi (GeoDataFrame | GeoSeries, optional): Area of interest for spatial filtering.
        stop_ids (list[str], optional): List of stop IDs to include.
    """
    if isinstance(path, (str, Path)):
        paths = [Path(path)]
    else:
        paths = [Path(p) for p in path]

    self.lf = self._read_stops(paths, stop_ids, check_files=check_files, min_file_id=min_file_id, collision_registry=collision_registry)

    if aoi is None:
        df = self.lf.select(
            ["stop_id", "parent_station", "stop_lat", "stop_lon"]
        ).collect()
        self.gdf = gpd.GeoDataFrame(
            {
                "stop_id": df["stop_id"],
                "parent_station": df["parent_station"],
            },
            geometry=gpd.points_from_xy(df["stop_lon"], df["stop_lat"]),
            crs="EPSG:4326",
        )
    else:
        self.lf = self.lf.collect().lazy()
        self.lf, self.gdf = self.filter_by_aoi(aoi)
        self.lf = self.lf.collect().lazy()

    if stop_group_distance > 0:
        self.lf = self.lf.collect().lazy()
        self.lf, self.gdf = self.group_stops(stop_group_distance)
        self.lf = self.lf.collect().lazy()

    if (aoi is not None) or (stop_group_distance > 0):
        self.stop_ids = self.lf.select("stop_id").collect()["stop_id"].to_list()
        if (len(self.stop_ids) > 0) and (self.stop_ids[0] is None):
            self.stop_ids = []
    else:
        self.stop_ids = None

    # Compute mean coordinates of stops (assumed self.stops.lf is LazyFrame)
    mean_coords = self.lf.select(
        [
            pl.col("stop_lon").mean().alias("mean_lon"),
            pl.col("stop_lat").mean().alias("mean_lat"),
        ]
    ).collect()

    self.mean_lon = mean_coords["mean_lon"][0]
    self.mean_lat = mean_coords["mean_lat"][0]

filter_by_aoi

filter_by_aoi(aoi: GeoDataFrame | GeoSeries) -> tuple[pl.LazyFrame, gpd.GeoDataFrame]

Filters stops by a given Area of Interest (AOI).

Performs: 1. Bounding box filter on LazyFrame (approximate). 2. Geometry-based intersection on GeoDataFrame (precise).

Parameters:

Name	Type	Description	Default
aoi	GeoDataFrame | GeoSeries	Area to filter stops within.	required

Returns:

Name	Type	Description
tuple	tuple[LazyFrame, GeoDataFrame]	(filtered LazyFrame, filtered GeoDataFrame)

Raises:

Type	Description
ValueError	If no stops are found within AOI.

Source code in pyGTFSHandler/models/stops.py

def filter_by_aoi(
    self, aoi: gpd.GeoDataFrame | gpd.GeoSeries
) -> tuple[pl.LazyFrame, gpd.GeoDataFrame]:
    """
    Filters stops by a given Area of Interest (AOI).

    Performs:
    1. Bounding box filter on LazyFrame (approximate).
    2. Geometry-based intersection on GeoDataFrame (precise).

    Args:
        aoi (GeoDataFrame | GeoSeries): Area to filter stops within.

    Returns:
        tuple: (filtered LazyFrame, filtered GeoDataFrame)

    Raises:
        ValueError: If no stops are found within AOI.
    """
    aoi = aoi.to_crs("EPSG:4326")
    minx, miny, maxx, maxy = aoi.total_bounds

    filtered_lf = self.lf.filter(
        (pl.col("stop_lon") > minx)
        & (pl.col("stop_lon") < maxx)
        & (pl.col("stop_lat") > miny)
        & (pl.col("stop_lat") < maxy)
    )

    df = filtered_lf.select(
        ["stop_id", "parent_station", "stop_lat", "stop_lon"]
    ).collect()

    gdf = gpd.GeoDataFrame(
        {
            "stop_id": df["stop_id"],
            "parent_station": df["parent_station"],
        },
        geometry=gpd.points_from_xy(df["stop_lon"], df["stop_lat"]),
        crs="EPSG:4326",
    )

    union_geom = aoi.union_all()
    gdf = gdf[gdf.intersects(union_geom)]

    if gdf.empty:
        raise ValueError("No stops found inside AOI bounds")

    stop_ids_df = pl.from_pandas(gdf[["stop_id"]]).lazy()
    final_lf = filtered_lf.join(stop_ids_df.lazy(), on=["stop_id"], how="semi")

    return final_lf, gdf

group_stops

group_stops(distance: float)

Groups nearby stops by spatial proximity and assigns consistent parent_station values.

Parameters:

Name	Type	Description	Default
distance	float	Max distance (in meters) for grouping.	required

Raises:

Type	Description
ValueError	If projection is missing or invalid.

Source code in pyGTFSHandler/models/stops.py

def group_stops(self, distance: float):
    """
    Groups nearby stops by spatial proximity and assigns consistent parent_station values.

    Args:
        distance (float): Max distance (in meters) for grouping.

    Raises:
        ValueError: If projection is missing or invalid.
    """
    # Polars-native clustering: bucket stops into a lon/lat grid sized to
    # `distance`, self-join each stop against its own cell plus the 8
    # neighboring cells to get cheap candidate pairs (never an all-pairs
    # cross join), keep pairs whose exact haversine distance is within
    # `distance`, and take connected components of that edge graph --
    # this is a *transitive* grouping (a chain of stops each within
    # `distance` of the next all end up in one cluster, even if the two
    # ends are farther apart than `distance`), unlike the "complete"
    # linkage clustering this replaces. No sklearn, no UTM projection.
    stop_coords = (
        self.lf.select(["stop_id", "stop_lat", "stop_lon"])
        .filter(
            pl.col("stop_id").is_not_null()
            & pl.col("stop_lat").is_not_null()
            & pl.col("stop_lon").is_not_null()
        )
        .unique("stop_id")
        .collect()
    )

    if os.environ.get("GTFS_STOP_GROUPING", "transitive").lower() == "complete":
        declared = (
            self.lf.select(["stop_id", "parent_station"]).unique("stop_id").collect()
            if "parent_station" in self.lf.collect_schema().names()
            else None
        )
        if declared is not None:
            stop_coords = stop_coords.join(declared, on="stop_id", how="left")
    cluster_labels = self._cluster_by_distance(stop_coords, distance)

    # This assumes parent_station info comes from self.lf (not self.gdf)
    parent_station_df = self.lf.select(["stop_id", "parent_station"])
    stop_ids_df = cluster_labels.lazy()

    cluster_df = stop_ids_df.join(parent_station_df, on=["stop_id"], how="left")

    # Merge clusters with common parent_station
    cluster_df = cluster_df.join(
        cluster_df.filter(
            pl.col("parent_station").is_not_null()
            & (pl.col("parent_station") != "")
        )
        .group_by("parent_station")
        .agg(pl.col("cluster").min().alias("merged_cluster")),
        on="parent_station",
        how="left",
    )

    cluster_df = cluster_df.with_columns(
        [
            pl.when(pl.col("merged_cluster").is_not_null())
            .then(pl.col("merged_cluster"))
            .otherwise(pl.col("cluster"))
            .alias("final_cluster"),
            pl.when(
                pl.col("parent_station").is_not_null()
                & (pl.col("parent_station") != "")
            )
            .then(pl.col("parent_station"))
            .otherwise(pl.col("stop_id"))
            .alias("fallback_id"),
        ]
    ).drop(["cluster", "merged_cluster"])

    # Assign one parent_station per cluster
    cluster_df = (
        cluster_df.group_by("final_cluster")
        .agg(
            [
                pl.col("stop_id"),
                pl.col("fallback_id").first().alias("parent_station"),
            ]
        )
        .with_columns(
            pl.col("parent_station")
            .cum_count()
            .over("parent_station")
            .alias("suffix_count"),
            pl.col("parent_station").is_duplicated().alias("is_dup"),
        )
        .explode(["stop_id"])
        .with_columns(
            (
                pl.when(pl.col("is_dup"))
                .then(
                    pl.col("parent_station")
                    + "_duplicated_"
                    + pl.col("suffix_count").cast(pl.Utf8)
                )
                .otherwise(pl.col("parent_station"))
            ).alias("parent_station")
        )
        .drop(["suffix_count", "is_dup", "final_cluster"])
        .with_columns(
            pl.when(
                pl.col("parent_station").is_null()
            ).then(
                pl.lit(None))
            .otherwise(
                pl.col("parent_station")
            ).alias("parent_station")
        )
        .with_columns(
            pl.when(
                pl.col("parent_station").is_null()
            ).then(
                pl.col("stop_id"))
            .otherwise(
                pl.col("parent_station")
            ).alias("parent_station")
        )
    )

    cluster_df = cluster_df.collect()

    # Update gdf and lf
    gdf = self.gdf.drop(columns=["parent_station"]).merge(
        cluster_df.to_pandas(), on=["stop_id"]
    )
    lf = self.lf.drop("parent_station").join(
        cluster_df.lazy(), on=["stop_id"], how="left"
    )

    return lf, gdf

reload_stops_lf

reload_stops_lf(path: Union[str, Path, List[Union[str, Path]]], stop_ids: Optional[LazyFrame] = None, collision_registry=None) -> None

Re-reads stops.txt from path and replaces self.lf.

Used by Feed.load after stop_times.txt has been loaded, so that stops referenced by in-scope stop_times but dropped by the earlier AOI/stop_id filtering (e.g. a bordering trip's next stop just outside the AOI) are brought back into self.lf.

Parameters:

Name	Type	Description	Default
path	Union[str, Path, List[Union[str, Path]]]	One or more GTFS directory paths (str, Path, or a list of either) to search for stops.txt.	required
stop_ids	Optional[LazyFrame]	Optional lazyframe/expression of stop_id values used to restrict which stops are (re)loaded. If None, all stops found in stops.txt are kept.	None

Returns:

Type	Description
None	None. self.lf is updated in place.

Source code in pyGTFSHandler/models/stops.py

def reload_stops_lf(
    self,
    path: Union[str, Path, List[Union[str, Path]]],
    stop_ids: Optional[pl.LazyFrame] = None,
    collision_registry=None,
) -> None:
    """Re-reads `stops.txt` from `path` and replaces `self.lf`.

    Used by `Feed.load` after `stop_times.txt` has been loaded, so that
    stops referenced by in-scope stop_times but dropped by the earlier
    AOI/stop_id filtering (e.g. a bordering trip's next stop just
    outside the AOI) are brought back into `self.lf`.

    Args:
        path: One or more GTFS directory paths (str, Path, or a list of
            either) to search for `stops.txt`.
        stop_ids: Optional lazyframe/expression of `stop_id` values used
            to restrict which stops are (re)loaded. If `None`, all
            stops found in `stops.txt` are kept.

    Returns:
        None. `self.lf` is updated in place.
    """
    if isinstance(path, (str, Path)):
        paths = [Path(path)]
    else:
        paths = [Path(p) for p in path]

    stop_paths: List[Path] = []
    file = "stops.txt"
    for p in paths:
        new_p = io.search_file(p, file=file)
        if new_p is None:
            stop_paths.append(None)
        else:
            stop_paths.append(new_p)

    schema_dict, _ = gtfs_checker.get_df_schema_dict("stops.txt")
    stops = io.read_csv_list(stop_paths, schema_overrides=schema_dict, check_files=True, collision_registry=collision_registry)
    stops = stops.filter(
        pl.col("stop_lat").is_not_null() & pl.col("stop_lon").is_not_null()
        & pl.col("stop_lat").is_finite() & pl.col("stop_lon").is_finite()
        & pl.col("stop_lat").is_between(-90, 90) & pl.col("stop_lon").is_between(-180, 180)
    )

    if isinstance(stop_ids, list):
        stop_ids_lf = pl.LazyFrame({"stop_id": stop_ids})

        # Select matching stop_times with just needed columns
        stops = stops.join(stop_ids_lf, on="stop_id", how="semi")
    elif stop_ids is not None:
        if isinstance(stop_ids, pl.DataFrame):
            stop_ids = stop_ids.lazy()

        columns = stop_ids.collect_schema().names()

        stops = stops.join(stop_ids, on=columns, how="semi")

    if "parent_station" in stops.collect_schema().names():
        stops = stops.with_columns(
            (
                pl.when(pl.col("parent_station").is_null())
                .then(pl.col("stop_id"))
                .otherwise(pl.col("parent_station"))
            ).alias("parent_station")
        )
    else:
        stops = stops.with_columns(pl.col("stop_id").alias("parent_station"))

    stops = (
        stops.join(
            self.lf.select("stop_id", "parent_station").rename(
                {"parent_station": "parent_station_right"}
            ),
            on=["stop_id"],
            how="left",
        )
        .with_columns(
            (
                pl.when(pl.col("parent_station_right").is_null())
                .then(pl.col("parent_station"))
                .otherwise(pl.col("parent_station_right"))
            ).alias("parent_station")
        )
        .drop("parent_station_right")
    )

    stops = stops.with_columns(
        pl.when(pl.col("parent_station") == "")
        .then(pl.lit(None))
        .otherwise("parent_station")
        .alias("parent_station"),
        pl.when(pl.col("stop_id") == "")
        .then(pl.lit(None))
        .otherwise("stop_id")
        .alias("stop_id")
    ).with_columns(
        pl.when(pl.col("parent_station").is_null())
        .then(pl.col("stop_id"))
        .otherwise("parent_station")
        .alias("parent_station"),
        pl.when(pl.col("stop_id").is_null())
        .then(pl.col("parent_station"))
        .otherwise("stop_id")
        .alias("stop_id")
    )
    stops = stops.filter(pl.col("stop_id").is_not_null() & (pl.col("stop_id") != ""))
    stops = stops.filter(pl.col("stop_lat").is_not_null() & pl.col("stop_lon").is_not_null())

    self.lf = stops
    return None