1. Home
  2. GTFS Feed Architecture Fundamentals
  3. GTFS Validation Rules And Common Schema Errors
  4. How To Validate A GTFS Feed With Python

How to Validate a GTFS Feed with Python

The most reliable production approach is to orchestrate the MobilityData GTFS Validator via subprocess, parse its structured JSON report, and gracefully degrade to a pandas-based schema checker when a Java runtime is unavailable. GTFS validation extends far beyond CSV parsing — it requires cross-file referential integrity checks, business-rule enforcement, and strict type validation that no pure-Python CSV reader handles completely out of the box.

Why GTFS Validation Requires Cross-File Integrity Checks

A GTFS feed is a ZIP archive containing interrelated CSV files that map to transit domains: agencies, stops, routes, trips, stop times, and service calendars. The foundational relationships are described in the GTFS Feed Architecture & Fundamentals guide. Validation failures rarely stem from malformed CSV syntax alone. They typically arise from broken foreign keys — a trip_id in stop_times.txt that has no matching record in trips.txt, invalid WGS84 coordinates, or overlapping service calendars. Catching these violations at the ingestion boundary, before the feed reaches routing engines or passenger-facing applications, is the core value proposition of any validation layer.

The diagram below shows the two-path validation architecture this guide implements: the preferred Java-backed path on the left and the Python-native fallback on the right.

GTFS Validation Architecture A flowchart showing two validation paths. The primary path uses the MobilityData Java validator via subprocess to produce a structured JSON report. When Java is missing, a pandas-based fallback checks required files, column headers, and a basic foreign key between stop_times and trips. Download GTFS ZIP urllib / requests Java available? YES MobilityData Java Validator Parse report.json notices + severities NO pandas Fallback Schema Check Files · Headers · FK stop_times → trips Structured validation report

Root Cause: Why Pure-Python Checks Are Insufficient

Most transit data teams initially reach for a CSV reader or pandas.read_csv to check a feed. This catches encoding errors and missing columns but misses the violations that actually break downstream systems:

Referential integrity gaps. The GTFS relational model requires stop_id values in stop_times.txt to resolve to records in stops.txt, route_id in trips.txt to resolve to routes.txt, and service_id in trips.txt to resolve to either calendar.txt or calendar_dates.txt. A pure CSV reader loads each file in isolation and cannot detect cross-file orphans. The complete FK graph is documented in GTFS Validation Rules and Common Schema Errors.

Enumeration and range constraints. Fields like route_type accept only integer values from a closed set (0–7 per GTFS Schedule Reference). location_type must be 0–4. Coordinates must satisfy WGS84 bounds — agencies that mis-order latitude and longitude produce coordinates that pass CSV parsing but geocode to the ocean. For coordinate-specific issues, see Coordinate Reference Systems for Transit Data.

Extended time format. The GTFS spec intentionally allows arrival_time and departure_time in stop_times.txt to exceed 23:59:59. A value of 25:30:00 means 1:30 AM of the next calendar day. Python’s datetime.strptime raises a ValueError on these inputs; the spec-compliant validator handles them correctly.

Business rule complexity. The MobilityData validator encodes over 120 rules derived from the specification and real-world agency behavior. Replicating all of them in pure Python is impractical. The subprocess orchestration approach in this guide delegates that logic to the authoritative Java implementation while keeping Python as the control plane.

Production-Ready Python Implementation

The script below downloads a GTFS ZIP, invokes the MobilityData validator, parses the output, and falls back to a Python-native schema check if no Java runtime is found. It uses tempfile to avoid leaving extraction artifacts, structured logging for pipeline observability, and explicit type hints throughout.

python 
import os
import subprocess
import json
import zipfile
import logging
import tempfile
import urllib.request
import pandas as pd
from pathlib import Path
from typing import Dict, List, Any, Optional

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
logger = logging.getLogger("gtfs_validator")

# ── Configuration ─────────────────────────────────────────────────────────────
GTFS_URL = "https://transit.agency/gtfs-static.zip"
VALIDATOR_JAR = Path(os.getenv("GTFS_VALIDATOR_JAR", "gtfs-validator.jar"))

REQUIRED_FILES: set[str] = {
    "agency.txt", "stops.txt", "routes.txt",
    "trips.txt", "stop_times.txt",
}
CALENDAR_FILES: set[str] = {"calendar.txt", "calendar_dates.txt"}

REQUIRED_HEADERS: dict[str, list[str]] = {
    "agency.txt":     ["agency_id", "agency_name", "agency_url", "agency_timezone"],
    "stops.txt":      ["stop_id", "stop_lat", "stop_lon"],
    "routes.txt":     ["route_id", "agency_id", "route_short_name", "route_type"],
    "trips.txt":      ["route_id", "service_id", "trip_id"],
    "stop_times.txt": ["trip_id", "arrival_time", "departure_time", "stop_id", "stop_sequence"],
}

# ── Step 1: Download ───────────────────────────────────────────────────────────
def download_gtfs(url: str, dest: Path) -> Path:
    """Download a GTFS ZIP from url to dest. Returns dest on success."""
    dest.parent.mkdir(parents=True, exist_ok=True)
    logger.info("Downloading feed from %s", url)
    urllib.request.urlretrieve(url, dest)
    logger.info("Saved feed to %s (%.1f MB)", dest, dest.stat().st_size / 1_048_576)
    return dest

# ── Step 2: Java validator ─────────────────────────────────────────────────────
def run_mobility_validator(
    zip_path: Path,
    output_dir: Path,
    heap: str = "2g",
    timeout_sec: int = 300,
) -> Dict[str, Any]:
    """
    Invoke the official MobilityData GTFS Validator JAR and return the JSON report.

    The validator exits with a non-zero code when it finds errors — that is
    expected behaviour, not a script failure. FileNotFoundError signals a missing
    Java runtime; we return a sentinel dict the caller uses to trigger the fallback.
    """
    output_dir.mkdir(parents=True, exist_ok=True)
    cmd: list[str] = [
        "java", f"-Xmx{heap}", "-jar", str(VALIDATOR_JAR),
        "-i", str(zip_path),
        "-o", str(output_dir),
    ]
    logger.info("Running validator: %s", " ".join(cmd))

    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=timeout_sec,
        )
        if result.returncode not in (0, 1):
            # Exit code 1 = validation notices found (normal)
            # Any other code = unexpected failure
            logger.warning(
                "Validator exited with code %d. stderr: %s",
                result.returncode,
                result.stderr[:400],
            )
    except subprocess.TimeoutExpired:
        logger.error("Validator timed out after %d seconds.", timeout_sec)
        return {"status": "timeout", "notices": []}
    except FileNotFoundError:
        logger.warning("Java runtime not found — switching to pandas fallback.")
        return {"status": "missing_runtime", "notices": []}

    report_path = output_dir / "report.json"
    if not report_path.exists():
        logger.error("No report.json produced. Check heap size or JAR path.")
        return {"status": "no_report", "notices": []}

    with report_path.open("r", encoding="utf-8") as fh:
        report = json.load(fh)

    report["status"] = "complete"
    return report

# ── Step 3: pandas fallback ────────────────────────────────────────────────────
def fallback_schema_check(zip_path: Path) -> Dict[str, Any]:
    """
    Python-native fallback: verifies required files, column headers, and the
    most critical foreign key (stop_times.trip_id → trips.trip_id).

    Sufficient for CI pre-checks but does not replace full spec validation.
    """
    findings: List[Dict[str, Any]] = []

    with tempfile.TemporaryDirectory() as tmpdir:
        tmp = Path(tmpdir)
        with zipfile.ZipFile(zip_path, "r") as zf:
            zf.extractall(tmp)

        present_files = {p.name for p in tmp.iterdir() if p.suffix == ".txt"}

        # ── Required file presence ─────────────────────────────────────────────
        missing = REQUIRED_FILES - present_files
        if missing:
            findings.append({
                "severity": "ERROR",
                "rule": "missing_required_files",
                "detail": sorted(missing),
            })
            return {"status": "fallback_complete", "findings": findings}

        if not (CALENDAR_FILES & present_files):
            findings.append({
                "severity": "ERROR",
                "rule": "missing_calendar",
                "detail": "Neither calendar.txt nor calendar_dates.txt present",
            })

        # ── Required column headers ────────────────────────────────────────────
        for fname, req_cols in REQUIRED_HEADERS.items():
            fpath = tmp / fname
            if not fpath.exists():
                continue
            try:
                df = pd.read_csv(
                    fpath,
                    dtype=str,
                    nrows=0,          # header only — fast
                    encoding="utf-8-sig",
                    keep_default_na=False,
                )
                missing_cols = set(req_cols) - set(df.columns)
                if missing_cols:
                    findings.append({
                        "severity": "ERROR",
                        "rule": "missing_columns",
                        "file": fname,
                        "detail": sorted(missing_cols),
                    })
            except pd.errors.ParserError as exc:
                findings.append({
                    "severity": "ERROR",
                    "rule": "parse_failure",
                    "file": fname,
                    "detail": str(exc),
                })

        # ── Foreign key check: stop_times.trip_id → trips.trip_id ─────────────
        try:
            trips_ids = pd.read_csv(
                tmp / "trips.txt",
                dtype=str,
                usecols=["trip_id"],
                encoding="utf-8-sig",
                keep_default_na=False,
            )["trip_id"].unique()

            st_ids = pd.read_csv(
                tmp / "stop_times.txt",
                dtype=str,
                usecols=["trip_id"],
                encoding="utf-8-sig",
                keep_default_na=False,
            )["trip_id"].unique()

            orphaned = set(st_ids) - set(trips_ids)
            if orphaned:
                findings.append({
                    "severity": "WARNING",
                    "rule": "orphaned_trip_ids",
                    "file": "stop_times.txt",
                    "orphan_count": len(orphaned),
                    "sample": sorted(orphaned)[:5],
                })
        except Exception as exc:
            findings.append({
                "severity": "WARNING",
                "rule": "fk_check_skipped",
                "detail": str(exc),
            })

    return {"status": "fallback_complete", "findings": findings}

# ── Step 4: Summarise report ───────────────────────────────────────────────────
def summarise_report(report: Dict[str, Any]) -> Dict[str, int]:
    """
    Count notices by severity from a MobilityData validator JSON report.
    Returns a dict with keys ERROR, WARNING, INFO, NOTE.
    """
    counts: dict[str, int] = {"ERROR": 0, "WARNING": 0, "INFO": 0, "NOTE": 0}
    for notice in report.get("notices", []):
        severity = notice.get("severity", "INFO").upper()
        if severity in counts:
            counts[severity] += 1
    return counts

# ── Entrypoint ─────────────────────────────────────────────────────────────────
def validate_feed(
    url: str,
    jar_path: Path = VALIDATOR_JAR,
    heap: str = "2g",
) -> Dict[str, Any]:
    zip_dest = Path("gtfs_feed.zip")
    output_dir = Path("validation_output")

    download_gtfs(url, zip_dest)

    report = run_mobility_validator(zip_dest, output_dir, heap=heap)
    if report.get("status") in ("missing_runtime", "timeout", "no_report"):
        logger.info("Switching to pandas fallback validation.")
        report = fallback_schema_check(zip_dest)

    zip_dest.unlink(missing_ok=True)
    return report


if __name__ == "__main__":
    result = validate_feed(GTFS_URL)
    print(json.dumps(result, indent=2, default=str))

Step-by-Step Walkthrough

Download and file management

download_gtfs writes the ZIP to a local path and logs the file size before returning. Logging the size immediately surfaces feeds that are suspiciously small (an agency CDN returning an HTML error page, for instance) before the validator wastes 300 seconds on a non-archive.

Subprocess invocation and exit-code semantics

run_mobility_validator wraps three distinct failure modes:

  • FileNotFoundError — the java binary is not on PATH. Return the missing_runtime sentinel; the caller routes to the fallback.
  • subprocess.TimeoutExpired — the feed is unusually large or the JVM is thrashing swap. Return timeout; do not block the pipeline indefinitely.
  • Non-zero exit with code ≠ 1 — the validator itself failed (corrupt JAR, out-of-memory, missing output directory). Log stderr and return no_report.

Exit code 1 is deliberately not treated as an error. The MobilityData validator exits 1 whenever it finds validation notices, which is the normal case for any real agency feed. Treating it as a subprocess failure would mask every result.

Reading report.json

The validator writes its output to the directory you pass with -o. The primary artifact is report.json, a structured document containing a notices array. Each entry carries severity, noticeCode, and a sampleMessages list. The summarise_report function provides a quick severity histogram for pipeline gating.

The pandas fallback

When Java is absent, fallback_schema_check performs three checks in order:

  1. File presence — verifies all five mandatory files and at least one calendar file exist. Failing here short-circuits the rest: downstream column checks on absent files produce meaningless results.
  2. Column headers — reads only the first row (via nrows=0) to check column names without loading data. This completes in under a second even for feeds where stop_times.txt has ten million rows.
  3. stop_times.trip_id → trips.trip_id FK — the single most impactful referential violation. Loading only the trip_id column from each file keeps memory usage proportional to the number of unique trips, not the number of stop time records. For memory-efficient processing of large feeds, usecols is non-negotiable.

The encoding="utf-8-sig" parameter throughout strips the UTF-8 BOM silently added by Windows-based publisher tools — a common source of phantom column-name mismatches.

Verification and Output

After running validate_feed, confirm the report is actionable before routing it downstream:

python 
def assert_report_usable(report: Dict[str, Any]) -> None:
    """
    Gate check: raise AssertionError if the report cannot be acted on.
    Call this before writing the report to your alert/monitoring system.
    """
    assert "status" in report, "Report missing 'status' field"
    assert report["status"] not in ("timeout",), (
        f"Validation did not complete: status={report['status']}"
    )

    if report["status"] == "complete":
        counts = summarise_report(report)
        logger.info(
            "Validation complete — ERROR: %d  WARNING: %d  INFO: %d  NOTE: %d",
            counts["ERROR"], counts["WARNING"], counts["INFO"], counts["NOTE"],
        )
        # Gate: fail the pipeline if structural errors are present
        assert counts["ERROR"] == 0, (
            f"Feed failed validation with {counts['ERROR']} ERROR(s). "
            "Inspect validation_output/report.json for details."
        )

    elif report["status"] == "fallback_complete":
        error_findings = [
            f for f in report.get("findings", [])
            if f.get("severity") == "ERROR"
        ]
        assert not error_findings, (
            f"Fallback check found {len(error_findings)} ERROR(s): "
            + "; ".join(f.get("rule", "unknown") for f in error_findings)
        )

Sample output from a clean MobilityData validator run:

json 
{
  "status": "complete",
  "notices": [
    {
      "severity": "WARNING",
      "noticeCode": "RouteShortNameTooLongNotice",
      "totalNotices": 3,
      "sampleMessages": [
        { "routeId": "B46", "routeShortName": "B46-SBS" }
      ]
    }
  ]
}

Sample output from the fallback path with an orphaned FK:

json 
{
  "status": "fallback_complete",
  "findings": [
    {
      "severity": "WARNING",
      "rule": "orphaned_trip_ids",
      "file": "stop_times.txt",
      "orphan_count": 14,
      "sample": ["BX12_LOCAL_0", "BX12_LOCAL_1", "M15_SBS_3"]
    }
  ]
}

Gotchas and Edge Cases

  • Heap sizing and silent failures. If the JVM runs out of heap, the validator process terminates without writing report.json. The script detects the missing file and returns no_report, but the root cause is often a feed that is larger than anticipated. Scale -Xmx to at least 4 GB for metro-area feeds. Pair this with a check of the ZIP file size before invoking the JAR.

  • calendar_dates.txt-only feeds. The GTFS spec permits feeds that omit calendar.txt entirely and express all service dates via calendar_dates.txt. The fallback’s file-presence check accounts for this by requiring at least one of the two calendar files. The Java validator handles this correctly by design. Do not hard-require calendar.txt in your own schema checks. Relevant context: Timezone Handling and Schedule Normalization covers how service-date semantics interact with DST boundaries.

  • Nested ZIP layouts. Some agencies publish archives with a subdirectory inside the ZIP (e.g., gtfs/stops.txt instead of stops.txt at the root). The Java validator rejects these outright. The fallback’s z.extractall(tmpdir) recreates the subdirectory, so the subsequent tmp.iterdir() call finds no .txt files and flags every mandatory file as missing. Add a flattening step — (tmp / fname).rename(tmp / Path(fname).name) for each nested path — if you routinely receive these feeds from specific agencies.

  • Mixed stop_id types across files. An agency may publish numeric stop_id values in stops.txt and zero-padded string equivalents in stop_times.txt (e.g., 1001 versus 01001). The dtype=str declarations in the fallback prevent silent integer coercion, but the FK check will still report these as orphans even though a human reader would consider them matching. Log sample values and compare against known agency patterns before treating this as a hard error.

FAQ: Validating GTFS Feeds with Python

Do I need Java installed to validate a GTFS feed with Python? The MobilityData GTFS Validator requires a Java 11+ runtime. When Java is unavailable, the pandas fallback in this guide verifies required files, column headers, and stop_times→trips referential integrity — sufficient for CI pre-checks but not for full spec compliance.

How do I parse the MobilityData validator JSON report in Python? The validator writes report.json to the output directory you specify with the -o flag. Load it with json.load() and iterate over the notices list. Each notice has a severity field (ERROR, WARNING, INFO, NOTE) and a noticeCode that maps to a spec rule.

What heap size should I pass to the GTFS validator JVM? Pass -Xmx2g for feeds under 500 MB. For large metropolitan feeds or multi-agency aggregates that exceed 1 GB uncompressed, use -Xmx4g. Insufficient heap causes the validator to silently exit without writing a report.

Up: GTFS Validation Rules and Common Schema Errors | GTFS Feed Architecture & Fundamentals | Home