Calendar

Calendar

Calendar(lf: Optional[LazyFrame] = None, exceptions_lf: Optional[LazyFrame] = None, min_date: Optional[int] = None, max_date: Optional[int] = None, service_ids: Optional[List[str]] = None)

Represents calendar.txt + calendar_dates.txt for one or more GTFS feeds.

Holds the raw (unmodified) weekday/date-range rows (self.lf) and the single-date exceptions (self.exceptions_lf), plus the resolved min_date/max_date validity window and, once load/filtering has run, the service_ids in scope. See the module docstring for how overnight trips are handled without duplicating rows.

Attributes:

Name	Type	Description
lf	LazyFrame | None	Raw calendar.txt rows.
exceptions_lf	LazyFrame | None	Raw calendar_dates.txt rows.
min_date	int | None	Earliest date covered, as days since epoch.
max_date	int | None	Latest date covered, as days since epoch.
service_ids	List[str] | None	service_ids in scope, if filtered.

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

Parameters:

Name	Type	Description	Default
lf	Optional[LazyFrame]	Optional pre-loaded calendar.txt LazyFrame (e.g. when reconstructing a Calendar from concat_feeds).	None
exceptions_lf	Optional[LazyFrame]	Optional pre-loaded calendar_dates.txt LazyFrame.	None
min_date	Optional[int]	Optional pre-computed min date (days since epoch); if omitted and lf/exceptions_lf are given, it is derived via _get_min_max_dates.	None
max_date	Optional[int]	Optional pre-computed max date (days since epoch).	None
service_ids	Optional[List[str]]	Optional pre-computed list of in-scope service_ids.	None

Source code in pyGTFSHandler/models/calendar.py

def __init__(
    self,
    lf: Optional[pl.LazyFrame] = None,
    exceptions_lf: Optional[pl.LazyFrame] = None,
    min_date: Optional[int] = None,
    max_date: Optional[int] = None,
    service_ids: Optional[List[str]] = None,
) -> None:
    """Wraps already-loaded calendar frames, or leaves them empty for `load` to fill in.

    Args:
        lf: Optional pre-loaded `calendar.txt` LazyFrame (e.g. when
            reconstructing a `Calendar` from `concat_feeds`).
        exceptions_lf: Optional pre-loaded `calendar_dates.txt` LazyFrame.
        min_date: Optional pre-computed min date (days since epoch);
            if omitted and `lf`/`exceptions_lf` are given, it is derived
            via `_get_min_max_dates`.
        max_date: Optional pre-computed max date (days since epoch).
        service_ids: Optional pre-computed list of in-scope `service_id`s.
    """
    self.lf = lf
    self.exceptions_lf = exceptions_lf 
    if (lf is not None) or (exceptions_lf is not None):
        if min_date is None:
            min_date, max_date = self._get_min_max_dates(
                lf, exceptions_lf
            )

    self.min_date = min_date 
    self.max_date = max_date 
    self.service_ids = service_ids

load

load(path: Union[str, Path, List[str], List[Path]], start_date: datetime | date | None = None, end_date: datetime | date | None = None, date_type: list[str] | str | None = None, lon: float | None = None, lat: float | None = None, service_ids: Optional[List[str]] | None = None, check_files: bool = False, min_file_id=0, collision_registry=None)

A class to manage GTFS calendar data, allowing filtering of active services by date using calendar.txt and calendar_dates.txt files.

Parameters:

Name	Type	Description	Default
path	str | Path | list[str | Path]	File or directory path(s) containing GTFS calendar files.	required
service_ids	list[str]	List of service IDs to filter on.	None

Attributes: path (Path): Directory path containing GTFS calendar files. lf (pl.LazyFrame | None): LazyFrame for calendar.txt data. exceptions_lf (pl.LazyFrame | None): LazyFrame for calendar_dates.txt data.

Source code in pyGTFSHandler/models/calendar.py

def load(
    self,
    path: Union[str, Path, List[str], List[Path]],
    start_date: datetime | date | None = None,
    end_date: datetime | date | None = None,
    date_type: list[str] | str | None = None,
    lon: float | None = None,
    lat: float | None = None,
    service_ids: Optional[List[str]] | None = None,
    check_files:bool=False,
    min_file_id=0,
    collision_registry=None,
):
    """
    A class to manage GTFS calendar data, allowing filtering of active services
    by date using calendar.txt and calendar_dates.txt files.


    Args:
        path (str | Path | list[str | Path]): File or directory path(s) containing GTFS calendar files.
        service_ids (list[str], optional): List of service IDs to filter on.


    Attributes:
    path (Path): Directory path containing GTFS calendar files.
    lf (pl.LazyFrame | None): LazyFrame for calendar.txt data.
    exceptions_lf (pl.LazyFrame | None): LazyFrame for calendar_dates.txt data.
    """
    # Normalize to list of Path
    if isinstance(path, (str, Path)):
        paths = [Path(path)]
    else:
        paths = [Path(p) for p in path]

    if isinstance(start_date, datetime):
        start_date = start_date.date()
    if isinstance(end_date, datetime):
        end_date = end_date.date()

    self.lf = self._read_calendar(paths, service_ids, check_files=check_files, min_file_id=min_file_id, collision_registry=collision_registry)
    self.exceptions_lf = self._read_calendar_dates(paths, service_ids, check_files=check_files, min_file_id=min_file_id, collision_registry=collision_registry)
    if (self.lf is None) and (self.exceptions_lf is None):
        raise Exception(f"No calendar.txt or calendar_dates.txt files found in paths {paths}")

    self.min_date, self.max_date = self._get_min_max_dates(
        self.lf, self.exceptions_lf
    )
    if start_date is not None and start_date > self.min_date:
        self.min_date = start_date
    if end_date is not None and end_date < self.max_date:
        self.max_date = end_date

    if isinstance(start_date, datetime):
        start_date = start_date.date()

    if isinstance(end_date, datetime):
        end_date = end_date.date()

    if start_date and end_date:
        if start_date == end_date:
            self.service_ids = list(
                set(self.get_services_in_date(start_date))
                | set(self.get_services_in_date(start_date - timedelta(days=1)))
            )
            service_ids_df = pl.DataFrame({"service_id": self.service_ids})
        elif start_date > end_date:
            raise Exception("Start date happens after end date")
        else:
            service_ids_df = (
                self.get_services_in_date_range(
                    start_date - timedelta(days=1),
                    end_date,
                    date_type=date_type,
                    lon=lon,
                    lat=lat,
                )
                .select("service_ids")
                .rename({"service_ids": "service_id"})
                .explode("service_id")
                .unique()
            )
            self.service_ids = service_ids_df["service_id"].to_list()

        if (len(self.service_ids) > 0) and (self.service_ids[0] is None):
            self.service_ids = []

        if self.lf is not None:
            self.lf = self.lf.join(
                service_ids_df.lazy(), on="service_id", how="semi"
            )

        if self.exceptions_lf is not None:
            self.exceptions_lf = self.exceptions_lf.join(
                service_ids_df.lazy(), on="service_id", how="semi"
            )

    else:
        self.service_ids = service_ids

get_services_in_date

get_services_in_date(date: datetime | date) -> List[str]

Returns a list of service_ids active on a given date.

Combines data from calendar.txt and calendar_dates.txt to include exceptions.

Parameters:

Name	Type	Description	Default
date	datetim | date	Date to check for active services.	required

Returns:

Type	Description
List[str]	list[str]: Sorted list of active service IDs on the given date.

Source code in pyGTFSHandler/models/calendar.py

def get_services_in_date(self, date: datetime | date) -> List[str]:
    """
    Returns a list of service_ids active on a given date.

    Combines data from calendar.txt and calendar_dates.txt to include
    exceptions.

    Args:
        date (datetim | date): Date to check for active services.

    Returns:
        list[str]: Sorted list of active service IDs on the given date.
    """
    date_int = date_parsing.datetime_to_days_since_epoch(date)
    weekday = date.strftime("%A").lower()  # e.g., 'monday'

    # Filter calendar.txt for services active on this weekday and date
    calendar_filtered = None
    if self.lf is not None:
        calendar_filtered = self.lf.filter(
            (pl.col(weekday) == 1)
            & (pl.col("start_date") <= date_int)
            & (pl.col("end_date") >= date_int)
        ).select("service_id")

    # Filter calendar_dates.txt for exceptions on this date
    remove_services = None
    add_services = None
    if self.exceptions_lf is not None:
        remove_services = self.exceptions_lf.filter(
            (pl.col("date") == date_int)
            & (pl.col("exception_type") == 2)  # Removed service
        ).select("service_id")

        add_services = self.exceptions_lf.filter(
            (pl.col("date") == date_int)
            & (pl.col("exception_type") == 1)  # Added service
        ).select("service_id")

    # Collect data frames as needed
    lazyframes = [
        df
        for df in [calendar_filtered, remove_services, add_services]
        if df is not None
    ]
    collected = pl.collect_all(lazyframes)

    calendar_df = (
        collected[0]
        if calendar_filtered is not None
        else pl.DataFrame({"service_id": []})
    )
    idx = 1 if calendar_filtered is not None else 0
    remove_df = (
        collected[idx]
        if remove_services is not None
        else pl.DataFrame({"service_id": []})
    )
    add_df = (
        collected[idx + 1]
        if add_services is not None
        else pl.DataFrame({"service_id": []})
    )

    # Compute final active service IDs
    active_services = (
        set(calendar_df["service_id"])
        .union(add_df["service_id"])
        .difference(remove_df["service_id"])
    )

    return list(active_services)

get_services_in_date_range

get_services_in_date_range(start_date: Optional[datetime | date] = None, end_date: Optional[datetime | date] = None, date_type: Optional[str | list[str]] = None, lon: float = None, lat: float = None) -> pl.DataFrame

Returns a Polars DataFrame with active service IDs for each date in the range.

The output contains columns

• 'date' (YYYY-MM-DD string)
• 'weekday' (lowercase weekday name)
• 'service_ids' (sorted list of active service IDs on that date)

If start_date or end_date is not provided, they are inferred from available data.

Parameters:

Name	Type	Description	Default
start_date	datetime | date	Start date of range.	None
end_date	datetime | date	End date of range.	None

Returns:

Type	Description
DataFrame	pl.DataFrame: DataFrame with active services per date.

Source code in pyGTFSHandler/models/calendar.py

def get_services_in_date_range(
    self,
    start_date: Optional[datetime | date] = None,
    end_date: Optional[datetime | date] = None,
    date_type: Optional[str | list[str]] = None,
    lon: float = None,
    lat: float = None,
) -> pl.DataFrame:
    """
    Returns a Polars DataFrame with active service IDs for each date in the range.

    The output contains columns:
        - 'date' (YYYY-MM-DD string)
        - 'weekday' (lowercase weekday name)
        - 'service_ids' (sorted list of active service IDs on that date)

    If start_date or end_date is not provided, they are inferred from available data.

    Args:
        start_date (datetime|date, optional): Start date of range.
        end_date (datetime|date, optional): End date of range.

    Returns:
        pl.DataFrame: DataFrame with active services per date.
    """

    # Use provided or inferred dates
    start_date = start_date or (self.min_date)
    end_date = end_date or (self.max_date)

    if isinstance(start_date, datetime):
        start_date = start_date.date()

    if isinstance(end_date, datetime):
        end_date = end_date.date()

    if end_date < start_date:
        raise Exception("end_date should be after start_date")

    # Generate list of dates in range with weekday info
    date_list = [
        start_date + timedelta(days=i)
        for i in range((end_date - start_date).days + 1)
    ]
    date_info = [
        {"date": d.isoformat(), "weekday": d.strftime("%A").lower(), "date_int": int(date_parsing.datetime_to_days_since_epoch(d))}
        for d in date_list
    ]

    # Map weekday to service IDs based on calendar.txt
    weekday_service_map = {
        wd: set()
        for wd in [
            "monday",
            "tuesday",
            "wednesday",
            "thursday",
            "friday",
            "saturday",
            "sunday",
        ]
    }

    if self.lf is not None:
        start_int = date_parsing.datetime_to_days_since_epoch(start_date)
        end_int = date_parsing.datetime_to_days_since_epoch(end_date)

        calendar_df = (
            self.lf.select(
                [
                    "service_id",
                    "start_date",
                    "end_date",
                    "monday",
                    "tuesday",
                    "wednesday",
                    "thursday",
                    "friday",
                    "saturday",
                    "sunday",
                ]
            )
            .filter(
                (pl.col("start_date") <= end_int)
                & (pl.col("end_date") >= start_int)
            )
            .collect()
        )

        # Initialize date to services mapping based on weekday
        date_service_map = {
            entry["date"]: {
                "weekday": entry["weekday"],
                "services": set(
                    calendar_df.filter(
                        (pl.col("start_date") <= entry["date_int"])
                        & (pl.col("end_date") >= entry["date_int"]) 
                        & (pl.col(entry["weekday"]) == 1)
                    )["service_id"].to_list()
                ),
            }
            for entry in date_info
        }
    else:
        # Initialize date to services mapping based on weekday
        date_service_map = {
            entry["date"]: {
                "weekday": entry["weekday"],
                "services": set(),
            }
            for entry in date_info
        }

    # Apply exceptions from calendar_dates.txt
    if self.exceptions_lf is not None:
        start_int = date_parsing.datetime_to_days_since_epoch(start_date)
        end_int = date_parsing.datetime_to_days_since_epoch(end_date)

        calendar_dates_df = (
            self.exceptions_lf.select(["date", "service_id", "exception_type"])
            .filter((pl.col("date") >= start_int) & (pl.col("date") <= end_int))
            .collect()
        )

        for row in calendar_dates_df.iter_rows(named=True):
            date_str = (date_parsing.EPOCH + timedelta(days=row["date"])).isoformat()
            service_id = row["service_id"]
            exception = row["exception_type"]

            if date_str not in date_service_map:
                date_service_map[date_str] = {"weekday": None, "services": set()}

            if exception == 1:
                date_service_map[date_str]["services"].add(service_id)
            elif exception == 2:
                date_service_map[date_str]["services"].discard(service_id)

    # Prepare final DataFrame output
    result = pl.DataFrame(
        [
            {
                "date": date,
                "weekday": data["weekday"],
                "service_ids": sorted(data["services"]),
            }
            for date, data in sorted(date_service_map.items())
        ] 
    ).with_columns(
        pl.col("date")
        .str.strptime(pl.Datetime, "%Y-%m-%d")
        .cast(pl.Date)
        .alias("date")
    )

    if date_type is not None:
        result = self.filter_by_date_type(result, date_type, lon, lat)

    return result

filter_by_date_type

filter_by_date_type(result: DataFrame, date_type: str | list[str], lon: float | None, lat: float | None) -> pl.DataFrame

Filters a per-date DataFrame down to rows matching date_type.

Extracted out of get_services_in_date_range so Feed can reuse the exact same weekday/weekend/holiday classification logic against a different date column than the one the row was originally computed for. This matters for multi-day trips: a stop_time's real calendar date is service_date + day_offset, which can fall on a different weekday (or even cross into a holiday) than its nominal service_date. Feed._filter_by_date_range calls this with result["date"]/ result["weekday"] already replaced by the offset-adjusted real date before requesting a "weekend"/"holiday"/etc. classification, so the classification is always evaluated against the day a stop_time actually, physically happens on.

Parameters:

Name	Type	Description	Default
result	DataFrame	DataFrame with at least date (pl.Date) and weekday (lowercase weekday name) columns, one row per calendar date.	required
date_type	str | list[str]	One or more of VALID_DATE_TYPES.	required
lon	float | None	Longitude used to resolve the country/subdivision for holiday lookups (only needed if date_type requests holiday-aware filtering).	required
lat	float | None	Latitude, see lon.	required

Returns:

Type	Description
DataFrame	pl.DataFrame: result filtered to rows matching every requested
DataFrame	date_type (AND semantics across multiple values).

Source code in pyGTFSHandler/models/calendar.py

def filter_by_date_type(
    self,
    result: pl.DataFrame,
    date_type: str | list[str],
    lon: float | None,
    lat: float | None,
) -> pl.DataFrame:
    """Filters a per-date DataFrame down to rows matching `date_type`.

    Extracted out of `get_services_in_date_range` so `Feed` can reuse the
    exact same weekday/weekend/holiday classification logic against a
    *different* date column than the one the row was originally computed
    for. This matters for multi-day trips: a stop_time's real calendar
    date is `service_date + day_offset`, which can fall on a different
    weekday (or even cross into a holiday) than its nominal service_date.
    `Feed._filter_by_date_range` calls this with `result["date"]`/
    `result["weekday"]` already replaced by the *offset-adjusted* real
    date before requesting a "weekend"/"holiday"/etc. classification, so
    the classification is always evaluated against the day a stop_time
    actually, physically happens on.

    Args:
        result: DataFrame with at least `date` (pl.Date) and `weekday`
            (lowercase weekday name) columns, one row per calendar date.
        date_type: One or more of `VALID_DATE_TYPES`.
        lon: Longitude used to resolve the country/subdivision for
            holiday lookups (only needed if `date_type` requests
            holiday-aware filtering).
        lat: Latitude, see `lon`.

    Returns:
        pl.DataFrame: `result` filtered to rows matching every requested
        `date_type` (AND semantics across multiple values).
    """
    if isinstance(date_type, str):
        date_type = [date_type]

    date_type = [dt.lower() for dt in date_type]
    invalid = [dt for dt in date_type if dt not in self.VALID_DATE_TYPES]
    if invalid:
        raise Exception(f"Date type(s) not implemented: {invalid}")

    needs_holiday_lookup = (
        ("holiday" in date_type) or
        ("workday" in date_type) or
        ("businessday" in date_type) or
        ("non_workday" in date_type) or
        ("non_businessday" in date_type)
    )
    needs_weekend = (
        needs_holiday_lookup
        or ("weekend" in date_type)
        or ("non_weekday" in date_type)
        or ("weekday" in date_type)
    )
    if needs_weekend:
        result = self.add_holidays_and_weekends(
            result, lon, lat, needs_holiday=needs_holiday_lookup
        )

    # Apply filters one by one (AND logic)
    if ("workday" in date_type) or ("businessday" in date_type):
        result = result.filter(
            (~ pl.col("holiday"))
            & (~ pl.col("weekend"))
        )

    if "weekday" in date_type:
        result = result.filter(
            (~ pl.col("weekend"))
        )

    if ("non_workday" in date_type) or ("non_businessday" in date_type):
        result = result.filter(
            (pl.col("holiday")) | (pl.col("weekend"))
        )

    if "holiday" in date_type:
        result = result.filter(pl.col("holiday"))

    if ("weekend" in date_type) or ("non_weekday" in date_type):
        result = result.filter(pl.col("weekend"))

    for day in [
        "monday",
        "tuesday",
        "wednesday",
        "thursday",
        "friday",
        "saturday",
        "sunday",
    ]:
        if day in date_type:
            result = result.filter(pl.col("weekday") == day)

    if "holiday" in date_type:
        result = result.drop("holiday", "weekend")

    return result

add_holidays_and_weekends

add_holidays_and_weekends(data, lon, lat, needs_holiday: bool = True)

Adds weekend (always, purely from weekday, no network) and, when needs_holiday is True, holiday (requires an external country/subdivision lookup + holiday-calendar fetch via utils).

Split out so date_type filters that only need weekend (e.g. a plain "weekend" request) never trigger the network-dependent holiday lookup.

Source code in pyGTFSHandler/models/calendar.py

def add_holidays_and_weekends(self, data, lon, lat, needs_holiday: bool = True):
    """Adds `weekend` (always, purely from `weekday`, no network) and,
    when `needs_holiday` is True, `holiday` (requires an external
    country/subdivision lookup + holiday-calendar fetch via `utils`).

    Split out so date_type filters that only need `weekend` (e.g. a
    plain `"weekend"` request) never trigger the network-dependent
    holiday lookup.
    """
    # If LazyFrame, collect to DataFrame
    if isinstance(data, pl.LazyFrame):
        data = data.collect()

    # Weekend: check if weekday is Saturday or Sunday (no network needed).
    data = data.with_columns(
        pl.col("weekday").is_in(["saturday", "sunday"]).alias("weekend")
    )

    if not needs_holiday:
        return data

    # Determine country and subdivision (your utils function)
    country_code, subdivision_code = geocoding.get_country_region(lat, lon)

    # Extract unique years as list of ints
    years = data.select(pl.col("date").dt.year()).unique().to_series().to_list()

    # Collect holidays for each year
    holidays_df = [
        date_parsing.get_holidays(year, country_code, subdivision_code) for year in years
    ]
    holidays_df = pl.concat(holidays_df)

    # Convert holidays_df 'date' column (assumed days since epoch) to datetime (Polars Date)
    holidays_df = holidays_df.with_columns(
        (
            pl.lit(date_parsing.EPOCH).cast(pl.Date) + pl.duration(days=pl.col("date"))
        ).alias("date")
    )

    # Holiday: if 'date' is in holidays_df
    data = data.with_columns(
        pl.col("date").is_in(holidays_df.get_column("date")).alias("holiday")
    )

    return data

See also Calendar handling methodology for the underlying date-resolution math.