Stop Times

StopTimes

StopTimes(lf: Optional[LazyFrame] = None, frequencies: Optional[LazyFrame] = None, fixed_times: Optional[bool] = None, trips_lf: Optional[LazyFrame] = None)

Bases: FrequenciesMixin

Manages and processes GTFS stop_times.txt and frequencies.txt data.

This class provides a comprehensive pipeline for reading, cleaning, and enriching stop time data from GTFS feeds. It leverages Polars LazyFrames for efficient, memory-friendly processing.

Key Features

• Reads and combines data from multiple GTFS sources.
• Filters data by stop_ids, trip_ids, and time windows.
• Normalizes GTFS times to seconds within a 0-24 hour day, using a next_day flag for services crossing midnight.
• Corrects invalid or missing stop_sequence values.
• Interpolates missing arrival/departure times (a more complex interpolation can be done together with the shapes file in the feed class).
• Calculates cumulative and total travel times for each trip.
• Expands frequency-based services into explicit stop times, creating new trip records for analysis.
• Generates canonical shape IDs by grouping trips with identical stop patterns and similar travel times.

Attributes:

Name	Type	Description
paths	List[Path]	A list of Path objects to the GTFS directories.
lf	LazyFrame	The main LazyFrame containing the processed stop times data.
fixed_times	bool	A flag indicating if any stop times were interpolated.
frequencies	Optional[LazyFrame]	A LazyFrame for processed frequencies data, or None if not present.

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

Parameters:

Name	Type	Description	Default
lf	Optional[LazyFrame]	Optional pre-loaded, processed stop_times.txt LazyFrame.	None
frequencies	Optional[LazyFrame]	Optional pre-loaded, processed frequencies.txt LazyFrame, or None if the feed has no frequencies.	None
fixed_times	Optional[bool]	Optional flag recording whether any times were interpolated during loading.	None
trips_lf	Optional[LazyFrame]	Optional trips LazyFrame updated with any synthetic trips generated from frequencies.txt.	None

Source code in pyGTFSHandler/models/stop_times.py

def __init__(
    self,
    lf: Optional[pl.LazyFrame] = None,
    frequencies: Optional[pl.LazyFrame] = None,
    fixed_times: Optional[bool] = None,
    trips_lf: Optional[pl.LazyFrame] = None,
) -> None:
    """Wraps already-loaded stop_times/frequencies frames, or leaves them empty for `load` to fill in.

    Args:
        lf: Optional pre-loaded, processed `stop_times.txt` LazyFrame.
        frequencies: Optional pre-loaded, processed `frequencies.txt`
            LazyFrame, or `None` if the feed has no frequencies.
        fixed_times: Optional flag recording whether any times were
            interpolated during loading.
        trips_lf: Optional trips LazyFrame updated with any synthetic
            trips generated from `frequencies.txt`.
    """
    self.lf = lf
    self.frequencies = frequencies 
    self.fixed_times = fixed_times
    self.trips_lf = trips_lf

load

load(path: Union[str, Path, List[Union[str, Path]]], trips, start_time: Optional[datetime] = None, end_time: Optional[datetime] = None, stop_ids: Optional[List[str] | DataFrame | LazyFrame] = None, trip_ids: Optional[List[str]] = None, check_files: bool = False, min_file_id=0, collision_registry=None)

Initializes the StopTimes instance and runs the processing pipeline.

Parameters:

Name	Type	Description	Default
path	Union[str, Path, List[Union[str, Path]]]	A single path or a list of paths to GTFS directories.	required
start_time	Optional[datetime]	The start of a time window for initial filtering. If provided, end_time must also be provided.	None
end_time	Optional[datetime]	The end of a time window for initial filtering. If provided, start_time must also be provided.	None
stop_ids	Optional[List[str]]	A list of stop_ids to filter the data upon loading. Only stop times associated with these stops will be processed.	None
trip_ids	Optional[List[str]]	A list of trip_ids to filter the data upon loading. Only stop times associated with these trips will be processed.	None

Source code in pyGTFSHandler/models/stop_times.py

def load(
    self,
    path: Union[str, Path, List[Union[str, Path]]],
    trips,
    start_time: Optional[datetime] = None,
    end_time: Optional[datetime] = None,
    stop_ids: Optional[List[str] | pl.DataFrame | pl.LazyFrame] = None,
    trip_ids: Optional[List[str]] = None,
    check_files:bool=False,
    min_file_id=0,
    collision_registry=None,
):
    """
    Initializes the StopTimes instance and runs the processing pipeline.

    Args:
        path (Union[str, Path, List[Union[str, Path]]]):
            A single path or a list of paths to GTFS directories.
        start_time (Optional[datetime]):
            The start of a time window for initial filtering. If provided,
            `end_time` must also be provided.
        end_time (Optional[datetime]):
            The end of a time window for initial filtering. If provided,
            `start_time` must also be provided.
        stop_ids (Optional[List[str]]):
            A list of `stop_id`s to filter the data upon loading. Only stop times
            associated with these stops will be processed.
        trip_ids (Optional[List[str]]):
            A list of `trip_id`s to filter the data upon loading. Only stop times
            associated with these trips will be processed.
    """
    # Standardize input path(s) to a list of Path objects
    if isinstance(path, (str, Path)):
        paths: List[Path] = [Path(path)]
    else:
        paths: List[Path] = [Path(p) for p in path]

    # --- Main Processing Pipeline ---
    self.lf: pl.LazyFrame = self._read_stop_times(paths, trip_ids, check_files=check_files, min_file_id=min_file_id, collision_registry=collision_registry)

    # Referential integrity: drop stop_times.txt rows whose trip_id has
    # no matching row in trips.txt (a dangling reference) *before* any
    # frequency-driven trip_id expansion happens below -- doing this
    # check later, against the post-expansion trip_ids, would wrongly
    # flag every legitimate frequency-generated instance as orphaned.
    known_trip_ids = trips.select("trip_id").unique()
    orphaned_trip_ids = (
        self.lf.select("trip_id").unique().join(known_trip_ids, on="trip_id", how="anti").collect()
    )
    self.lf = self.lf.join(known_trip_ids, on="trip_id", how="semi")
    if orphaned_trip_ids.height > 0:
        warnings.warn(
            f"{orphaned_trip_ids.height} trip_id(s) in stop_times.txt "
            f"do not exist in trips.txt; their stop_times were dropped."
        )

    self.lf = self._correct_sequence(self.lf)
    self.lf, self.fixed_times = self._fix_nulls_easy(self.lf)
    self.lf = self._normalize_times(self.lf)

    if stop_ids is not None:
        self.lf = self._filter_by_stop_id(self.lf, stop_ids)
        self.lf = self._correct_sequence(self.lf)
        self.lf = self.lf.collect().lazy()

    if self.fixed_times:
        warnings.warn("Some departure times are null and have been interpolated")

    self.frequencies: Optional[pl.LazyFrame] = self._read_frequencies(
        paths, trip_ids, check_files=check_files, min_file_id=min_file_id, collision_registry=collision_registry
    )

    if self.frequencies is not None:
        self.lf = self.lf.collect().lazy()

        self.frequencies = self.frequencies.join(self.lf, on="trip_id", how="semi")
        if check_files:
            self.frequencies = self._filter_repeated_frequencies_with_trips(
                self.frequencies, trips
            )
        self.frequencies = self.frequencies.collect().lazy()
        self.frequencies = self._frequencies_midnight_crossing(self.frequencies)
        if check_files:
            self.lf, self.frequencies = self._check_frequencies_in_stop_times(
                self.lf, self.frequencies
            )

        self.lf = self.lf.collect().lazy()
        self.frequencies = self.frequencies.collect().lazy()

    self.lf = self._add_shape_time_and_midnight_crossing(self.lf)

    if self.frequencies is not None:
        self.frequencies = self._add_departure_time_to_frequencies(
            self.lf, self.frequencies
        )

    if start_time and end_time:
        self.lf = self.lf.collect().lazy()
        if self.frequencies is not None:
            self.frequencies = self.frequencies.collect().lazy()

        self.lf, self.frequencies = self.filter_by_time_range(
            start_time, end_time, strict=False
        )

        if self.frequencies is not None:
            self.frequencies = self.frequencies.collect().lazy()

    # Eagerly evaluate to checkpoint the result and optimize subsequent queries
    self.lf = self.lf.collect().lazy()

    if self.frequencies is not None:
        self.frequencies = self._fix_headway(self.frequencies)
        self.lf, self.frequencies = self._midnight_frequencies_to_stop_times(
            self.lf, self.frequencies
        )
        if start_time and end_time:
            self.frequencies = self.frequencies.filter(
                (pl.col("start_time") < time_parsing.time_to_seconds(end_time))
                & (pl.col("end_time") > time_parsing.time_to_seconds(start_time))
            )

        # Eagerly evaluate frequencies to checkpoint results
        self.frequencies = self.frequencies.collect().lazy()

        # Reconcile each trip's chain of windows against headway_secs
        # after every load-time filter (initial start/end window,
        # midnight-crossing split) has already shrunk/adjusted them --
        # always run, regardless of check_files, so no window's fractional
        # trailing period leaks into an adjacent window.
        self.frequencies = self._reconcile_frequency_windows(
            self.frequencies, check_files=check_files
        )
        self.frequencies = self.frequencies.collect().lazy()

        self.frequencies = self._add_frequencies_n_trips(self.frequencies)

        # Recalculate travel times as new trips may have been added
        self.lf = self._add_shape_time_and_midnight_crossing(self.lf)
        self.lf = self.lf.collect().lazy()

        if "orig_trip_id" in self.lf.collect_schema().names():
            unique_trip_ids: pl.LazyFrame = (
                self.lf.select(["trip_id", "orig_trip_id"])
                .unique()
                .rename({"trip_id": "new_trip_id"})
            )

            # Join the trips table with this mapping. This duplicates the original
            # trip's data (route_id, service_id, etc.) for each new generated trip_id.
            trips = (
                trips.join(
                    unique_trip_ids,
                    left_on="trip_id",
                    right_on="orig_trip_id",
                    how="right",  # Right join ensures all new trip_ids from stop_times are included.
                )
                .with_columns(pl.col("new_trip_id").alias("trip_id"))
                .drop("new_trip_id", "orig_trip_id")
            )

            self.lf = self.lf.drop("orig_trip_id")

    # `day_offset` on `self.lf` already resolves which real calendar day
    # each stop belongs to (see `_normalize_times`), so trips no longer
    # need to be duplicated as a separate "_night" service_id/trip_id
    # variant the way the old single-bit `next_day` scheme required.
    self.trips_lf = trips

generate_shape_ids

generate_shape_ids() -> pl.LazyFrame

Groups trips by stop sequence and travel time to create canonical shape IDs.

This method generates new shape_ids. A shape is defined as a group of trips that share the exact same sequence of stops and have a total travel duration within a configurable tolerance (TRIP_ROUND_TIME, e.g., 2 minutes).

This is useful for identifying all trips that follow the same physical path with a similar duration, effectively creating a shape_id when one is not provided or is inconsistent in the GTFS data.

Returns:

Type	Description
LazyFrame	pl.LazyFrame: A LazyFrame where each row represents a unique "shape", containing the generated shape_id, a list of all trip_ids belonging to it, the common stop_ids list, and the stop_sequence.

Source code in pyGTFSHandler/models/stop_times.py

def generate_shape_ids(self) -> pl.LazyFrame:
    """
    Groups trips by stop sequence and travel time to create canonical shape IDs.

    This method generates new `shape_id`s. A shape is defined as a group of trips
    that share the exact same sequence of stops and have a total travel duration
    within a configurable tolerance (`TRIP_ROUND_TIME`, e.g., 2 minutes).

    This is useful for identifying all trips that follow the same physical path
    with a similar duration, effectively creating a `shape_id` when one is not
    provided or is inconsistent in the GTFS data.

    Returns:
        pl.LazyFrame: A LazyFrame where each row represents a unique "shape",
                      containing the generated `shape_id`, a list of all `trip_ids`
                      belonging to it, the common `stop_ids` list, and the
                      `stop_sequence`.
    """
    trip_sequences = (
        self.lf.sort(["trip_id", "stop_sequence"])
        .group_by("trip_id")
        .agg(
            [
                pl.col("stop_id").sort_by("stop_sequence").alias("stop_ids"),
                pl.col("stop_sequence").sort().alias("stop_sequence"),
                pl.col("shape_total_travel_time")
                .first()
                .alias("shape_total_travel_time"),
                (
                    pl.col("shape_total_travel_time").first()
                    // TRIP_ROUND_TIME
                    * TRIP_ROUND_TIME
                ).alias("shape_total_travel_time_rounded"),
            ]
        )
    )

    grouped = trip_sequences.group_by(
        ["stop_ids", "shape_total_travel_time_rounded"]
    ).agg(
        [
            pl.col("trip_id").unique().alias("trip_ids"),
            pl.col("stop_sequence").first(),
            pl.col("trip_id").min().alias("shape_id"),
        ]
    )

    return grouped

filter_by_time_range

filter_by_time_range(start_time: datetime | time, end_time: datetime | time, strict: bool = True) -> Tuple[pl.LazyFrame, Optional[pl.LazyFrame]]

Filters stop_times and frequencies by a single time interval.

It handles intervals on the same date. If frequencies data is available, it ensures that entire trips belonging to an overlapping frequency definition are kept, even if not all their stops fall within the time bounds.

Parameters:

Name	Type	Description	Default
start_time	datetime | time	Start datetime of the filter interval.	required
end_time	datetime | time	End datetime of the filter interval.	required

Returns:

Type	Description
Tuple[LazyFrame, Optional[LazyFrame]]	Tuple[pl.LazyFrame, Optional[pl.LazyFrame]]: A tuple containing: - filtered_stop_times: LazyFrame filtered to the time interval. - filtered_frequencies: LazyFrame of frequencies overlapping the interval, or None if no frequency data exists.

Raises:

Type	Description
ValueError	If start_time and end_time are not on the same date.

Source code in pyGTFSHandler/models/stop_times.py

def filter_by_time_range(
    self,
    start_time: datetime | time,
    end_time: datetime | time,
    strict: bool = True,
) -> Tuple[pl.LazyFrame, Optional[pl.LazyFrame]]:
    """
    Filters stop_times and frequencies by a single time interval.

    It handles intervals on the same date. If `frequencies` data is available,
    it ensures that entire trips belonging to an overlapping frequency definition
    are kept, even if not all their stops fall within the time bounds.

    Args:
        start_time (datetime|time): Start datetime of the filter interval.
        end_time (datetime|time): End datetime of the filter interval.

    Returns:
        Tuple[pl.LazyFrame, Optional[pl.LazyFrame]]: A tuple containing:
            - `filtered_stop_times`: LazyFrame filtered to the time interval.
            - `filtered_frequencies`: LazyFrame of frequencies overlapping the interval,
              or None if no frequency data exists.

    Raises:
        ValueError: If `start_time` and `end_time` are not on the same date.
    """
    start_secs = time_parsing.time_to_seconds(start_time)
    end_secs = time_parsing.time_to_seconds(end_time)

    if self.frequencies is None:
        if (
            isinstance(start_time, datetime)
            and isinstance(end_time, datetime)
            and start_time.date() != end_time.date()
        ):
            raise ValueError("Start and end datetime must be on the same date")

        if strict:
            filtered_stop_times = self.lf.filter(
                (pl.col("arrival_time") >= start_secs)
                & (pl.col("arrival_time") <= end_secs)
            )
        else:
            filtered_stop_times = self.lf.join(
                self.lf.filter(
                    (pl.col("arrival_time") >= start_secs)
                    & (pl.col("arrival_time") <= end_secs)
                ),
                on="trip_id",
                how="semi",
            )

        return filtered_stop_times, None

    if strict:
        filtered_frequencies = self.frequencies.filter(
            (pl.col("start_time") < end_secs) & (pl.col("end_time") > start_secs)
        )

        filtered_frequencies = filtered_frequencies.with_columns(
            [
                pl.when(pl.col("end_time") > end_secs)
                .then(end_secs)
                .otherwise(pl.col("end_time"))
                .alias("end_time"),
                pl.when(pl.col("start_time") < start_secs)
                .then(start_secs)
                .otherwise(pl.col("start_time"))
                .alias("start_time"),
            ]
        )

    else:
        filtered_frequencies = self.frequencies.filter(
            (pl.col("start_time") < end_secs) & (pl.col("end_time") > start_secs)
        )

        filtered_frequencies = filtered_frequencies.with_columns(
            [
                pl.when(pl.col("end_time") > end_secs)
                .then(end_secs)
                .otherwise(pl.col("end_time"))
                .alias("end_time"),
                pl.when(
                    (pl.col("start_time") - pl.col("shape_total_travel_time"))
                    < (start_secs - pl.col("shape_total_travel_time"))
                )
                .then(
                    pl.when((start_secs - pl.col("shape_total_travel_time")) < 0)
                    .then(0)
                    .otherwise(start_secs - pl.col("shape_total_travel_time"))
                )
                .otherwise(pl.col("start_time"))
                .alias("start_time"),
            ]
        )

    if strict:
        filtered_stop_times = pl.concat(
            [
                self.lf.join(filtered_frequencies, on="trip_id", how="semi"),
                self.lf.join(filtered_frequencies, on="trip_id", how="anti").filter(
                    (pl.col("arrival_time") >= start_secs)
                    & (pl.col("arrival_time") <= end_secs)
                ),
            ]
        )
    else:
        valid_trip_ids = pl.concat(
            [
                filtered_frequencies.select("trip_id"),
                self.lf.join(filtered_frequencies, on="trip_id", how="anti")
                .filter(
                    (pl.col("arrival_time") >= start_secs)
                    & (pl.col("arrival_time") <= end_secs)
                )
                .select("trip_id"),
            ]
        )

        filtered_stop_times = self.lf.join(valid_trip_ids, on="trip_id", how="semi")

    filtered_stop_times = self._correct_sequence(filtered_stop_times)
    filtered_stop_times = self._add_shape_time_and_midnight_crossing(
        filtered_stop_times
    )

    return filtered_stop_times, filtered_frequencies

to_hhmmss

to_hhmmss(field: str, new_field: str) -> pl.Expr

Creates a Polars expression to convert seconds since midnight to a HH:MM:SS string.

Parameters:

Name	Type	Description	Default
field	str	The name of the column containing seconds (integer).	required
new_field	str	The desired name for the new HH:MM:SS string column.	required

Returns:

Type	Description
Expr	pl.Expr: A Polars expression that performs the conversion. Can be used in a .with_columns() statement.

Source code in pyGTFSHandler/models/stop_times.py

def to_hhmmss(self, field: str, new_field: str) -> pl.Expr:
    """
    Creates a Polars expression to convert seconds since midnight to a HH:MM:SS string.

    Args:
        field (str): The name of the column containing seconds (integer).
        new_field (str): The desired name for the new HH:MM:SS string column.

    Returns:
        pl.Expr: A Polars expression that performs the conversion. Can be used in
                 a `.with_columns()` statement.
    """
    seconds_expr = pl.col(field)
    hours = (seconds_expr // 3600).cast(pl.Int32)
    minutes = ((seconds_expr % 3600) // 60).cast(pl.Int32)
    seconds = (seconds_expr % 60).cast(pl.Int32)
    return (pl.format("{:02}:{:02}:{:02}", hours, minutes, seconds)).alias(
        new_field
    )