diff --git a/mobility/runtime/parameter_profiles.py b/mobility/runtime/parameter_profiles.py index 5bc398f9..a771becf 100644 --- a/mobility/runtime/parameter_profiles.py +++ b/mobility/runtime/parameter_profiles.py @@ -51,9 +51,9 @@ def at(self, iteration: int) -> float: class ListParameterProfile(ParameterProfile): """List-valued parameter profile supporting step-wise changes only.""" - points: dict[int, list[str]] + points: dict[int, list[Any]] - def at(self, iteration: int) -> list[str]: + def at(self, iteration: int) -> list[Any]: sorted_points = sorted(self.points.items()) iterations = [iteration for iteration, _ in sorted_points] idx = np.searchsorted(iterations, iteration, side="right") - 1 diff --git a/mobility/transport/modes/public_transport/gtfs/gtfs_edit.py b/mobility/transport/modes/public_transport/gtfs/gtfs_edit.py new file mode 100644 index 00000000..1784a05b --- /dev/null +++ b/mobility/transport/modes/public_transport/gtfs/gtfs_edit.py @@ -0,0 +1,383 @@ +"""Small GTFS edit helpers used to patch a feed before routing. + +The current feature set is intentionally narrow: +- insert a stop between two consecutive stops +- propagate the resulting delay before, after, or symmetrically +""" + +import hashlib +import logging +import math +import re +import zipfile +from dataclasses import dataclass +from io import BytesIO +from pathlib import Path + +import pandas as pd + + +@dataclass(frozen=True) +class NewStop: + stop_id: str + stop_name: str + stop_lat: float + stop_lon: float + + +class GTFSFeed: + def __init__(self, zip_path: str | Path): + self.zip_path = Path(zip_path) + self.tables: dict[str, pd.DataFrame] = {} + + def load(self) -> "GTFSFeed": + with zipfile.ZipFile(self.zip_path, "r") as z: + for name in z.namelist(): + if name.endswith(".txt"): + with z.open(name) as f: + self.tables[name] = pd.read_csv(f) + return self + + def save(self, out_zip_path: str | Path) -> Path: + out_zip_path = Path(out_zip_path) + buf = BytesIO() + with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as z: + for name, df in self.tables.items(): + z.writestr(name, df.to_csv(index=False)) + buf.seek(0) + out_zip_path.write_bytes(buf.read()) + return out_zip_path + + +def apply_gtfs_edits(gtfs_files, gtfs_edits, edits_folder: str | Path): + edits_folder = Path(edits_folder) + edits_folder.mkdir(parents=True, exist_ok=True) + + def matches_rule(rule, gtfs_path): + if rule.get("path"): + return str(Path(rule["path"])) == str(Path(gtfs_path)) + if rule.get("match"): + return rule["match"] in str(gtfs_path) + return False + + def core_id(x): + m = re.search(r"(\d{5,})$", str(x)) + return m.group(1) if m else str(x) + + def has_chain(gtfs_path, from_id, to_id): + with zipfile.ZipFile(gtfs_path, "r") as z: + with z.open("stop_times.txt") as f: + st = pd.read_csv(f, usecols=["trip_id", "stop_sequence", "stop_id"]) + st = st.sort_values(["trip_id", "stop_sequence"]) + st["a"] = st["stop_id"].map(core_id) + st["b"] = st.groupby("trip_id", sort=False)["a"].shift(-1) + return ((st["a"] == core_id(from_id)) & (st["b"] == core_id(to_id))).any() + + def expand_ops(rule_ops): + """Duplique les ops avec bidirectional=True en ajoutant l'op inversee.""" + expanded = [] + for op in rule_ops: + expanded.append(op) + if op.get("op") == "insert_stop_between" and op.get("bidirectional"): + op_rev = dict(op) + op_rev["from_stop_id"], op_rev["to_stop_id"] = op["to_stop_id"], op["from_stop_id"] + # On peut garder bidirectional=True ou le mettre a False pour eviter re-expansion. + op_rev["bidirectional"] = False + expanded.append(op_rev) + return expanded + + # --- Precompute chain hits for each (from,to) actually needed --- + chain_hits = {} + for rule in gtfs_edits or []: + for op in expand_ops(rule.get("ops", [])): + if op.get("op") != "insert_stop_between": + continue + key = (op["from_stop_id"], op["to_stop_id"]) + if key in chain_hits: + continue + + hits = [] + for p in gtfs_files: + try: + if has_chain(p, key[0], key[1]): + hits.append(p) + except Exception: + pass + chain_hits[key] = hits + + # --- Build mapping: gtfs_path -> list of ops to apply --- + ops_by_gtfs = {} + for rule in gtfs_edits or []: + mode = (rule.get("mode") or "explicit").lower().strip() + if mode not in {"explicit", "all"}: + raise ValueError("rule.mode must be 'explicit' or 'all'") + + for op in expand_ops(rule.get("ops", [])): + if op.get("op") != "insert_stop_between": + continue + + key = (op["from_stop_id"], op["to_stop_id"]) + hits = chain_hits.get(key, []) + + if mode == "explicit": + targets = [p for p in gtfs_files if matches_rule(rule, p)] + not_covered = [p for p in hits if p not in targets] + if not_covered: + logging.info( + "[GTFS edit] Note: chain %s -> %s also found in %s GTFS not targeted (mode=explicit).", + key[0], + key[1], + len(not_covered), + ) + for p in not_covered: + logging.info("[GTFS edit] - %s", p) + else: + targets = hits + logging.info( + "[GTFS edit] mode=all: applying chain %s -> %s edit to %s GTFS.", + key[0], + key[1], + len(targets), + ) + + for p in targets: + ops_by_gtfs.setdefault(p, []).append(op) + + # --- Apply edits (inchange) --- + new_files = [] + for gtfs_path in gtfs_files: + ops = ops_by_gtfs.get(gtfs_path) + if not ops: + new_files.append(gtfs_path) + continue + + h = hashlib.md5((str(gtfs_path) + str(ops)).encode("utf-8")).hexdigest()[:12] + src_p = Path(gtfs_path) + out_p = edits_folder / f"{src_p.stem}__edited_{h}{src_p.suffix}" + + if out_p.exists(): + logging.info("[GTFS edit] Using cached edited GTFS: %s", out_p) + new_files.append(str(out_p)) + continue + + logging.info("[GTFS edit] Editing GTFS: %s", src_p.name) + feed = GTFSFeed(src_p).load() + + for op in ops: + ns = op["new_stop"] + insert_stop_between( + feed=feed, + from_stop_id=op["from_stop_id"], + to_stop_id=op["to_stop_id"], + new_stop=NewStop( + stop_id=ns["stop_id"], + stop_name=ns["stop_name"], + stop_lat=ns["stop_lat"], + stop_lon=ns["stop_lon"], + ), + dwell_time_s=int(op.get("dwell_time_s", 45)), + extra_run_time_s=int(op.get("extra_run_time_s", 30)), + split_ratio=float(op.get("split_ratio", 0.5)), + propagate=str(op.get("propagate", "after")), + ) + + feed.save(out_p) + logging.info("[GTFS edit] Saved edited GTFS: %s", out_p) + new_files.append(str(out_p)) + + return new_files + + +def insert_stop_between( + feed: GTFSFeed, + from_stop_id: str, + to_stop_id: str, + new_stop: NewStop, + dwell_time_s: int = 45, + extra_run_time_s: int = 30, + split_ratio: float = 0.5, + propagate: str = "after", # "after" | "before" | "symmetric" +) -> None: + """ + Insert a new stop between two consecutive stops A -> B for all trips. + + Assumptions (heuristic, not operationally exact): + - We split the original A->B travel time using split_ratio to position the new stop temporally. + - We add dwell_time_s at the new stop. + - We add extra_run_time_s to represent braking/acceleration overhead. + - We propagate the resulting delay either after B (default), before A, or symmetrically. + + Matching: + - We match stops by their numeric suffix (e.g. "...87118257"), to ignore StopPoint/StopArea prefixes. + """ + propagate = (propagate or "after").lower().strip() + if propagate not in {"after", "before", "symmetric"}: + raise ValueError("propagate must be one of: after, before, symmetric") + if not (0.0 <= float(split_ratio) <= 1.0): + raise ValueError("split_ratio must be between 0 and 1") + + logging.info( + "[GTFS edit] insert_stop_between: start (from=%s, to=%s, new=%s)", + from_stop_id, + to_stop_id, + new_stop.stop_id, + ) + + # --- Inline helpers (kept minimal) --- + id_re = re.compile(r"(\d{5,})$") + + def core_id(x: str) -> str: + if not isinstance(x, str): + return "" + m = id_re.search(x) + return m.group(1) if m else x + + def hms_to_s(x: str): + # Returns int seconds or NA. + if not isinstance(x, str) or x == "": + return pd.NA + try: + h, m, s = x.split(":") + return int(h) * 3600 + int(m) * 60 + int(s) + except Exception: + return pd.NA + + def s_to_hms(x) -> str: + # GTFS allows hours > 24, keep as-is (no modulo). + if x is None or (isinstance(x, float) and math.isnan(x)) or pd.isna(x): + return "" + x = int(round(float(x))) + h = x // 3600 + m = (x % 3600) // 60 + s = x % 60 + return f"{h:02d}:{m:02d}:{s:02d}" + + from_core = core_id(from_stop_id) + to_core = core_id(to_stop_id) + + # --- stops.txt: minimal insert (do not try to rebuild full StopArea/StopPoint hierarchy) --- + if "stops.txt" not in feed.tables: + raise ValueError("stops.txt missing in GTFS feed") + if "stop_times.txt" not in feed.tables: + raise ValueError("stop_times.txt missing in GTFS feed") + + stops = feed.tables["stops.txt"] + if (stops["stop_id"] == new_stop.stop_id).sum() == 0: + row = { + "stop_id": new_stop.stop_id, + "stop_name": new_stop.stop_name, + "stop_lat": new_stop.stop_lat, + "stop_lon": new_stop.stop_lon, + } + # Ensure required cols exist; keep other cols untouched / NA. + for col in row.keys(): + if col not in stops.columns: + stops[col] = pd.NA + feed.tables["stops.txt"] = pd.concat([stops, pd.DataFrame([row])], ignore_index=True) + + # --- stop_times.txt: edit trip-by-trip --- + st = feed.tables["stop_times.txt"].copy() + st["arrival_s"] = pd.to_numeric(st["arrival_time"].map(hms_to_s), errors="coerce") + st["departure_s"] = pd.to_numeric(st["departure_time"].map(hms_to_s), errors="coerce") + st.sort_values(["trip_id", "stop_sequence"], inplace=True) + + # We accumulate "insertions" + "shifts" and apply them once at the end. + insert_rows = [] + seq_shifts = [] # (trip_id, start_seq, +1) + time_shifts = [] # (trip_id, mode, boundary_seq, seconds) + + modified_trips = set() + delta_total = int(dwell_time_s) + int(extra_run_time_s) + + for trip_id, g in st.groupby("trip_id", sort=False): + g = g.sort_values("stop_sequence") + + seq = g["stop_sequence"].to_list() + stop_ids = g["stop_id"].to_list() + stop_core = [core_id(s) for s in stop_ids] + + for i in range(len(stop_core) - 1): + if stop_core[i] != from_core or stop_core[i + 1] != to_core: + continue + + dep_a = g.iloc[i]["departure_s"] + arr_b = g.iloc[i + 1]["arrival_s"] + if pd.isna(dep_a) or pd.isna(arr_b): + # If times are missing, we skip this occurrence (keeps edit safe). + continue + + seq_a = int(seq[i]) + seq_b = int(seq[i + 1]) + + dep_a = int(round(float(dep_a))) + arr_b = int(round(float(arr_b))) + + # Original travel time between A departure and B arrival. + t_ab = max(0, arr_b - dep_a) + + # Place the new stop along A->B time axis. + t_an = int(t_ab * float(split_ratio)) + + arr_n = dep_a + t_an + dep_n = arr_n + int(dwell_time_s) + + insert_rows.append( + { + "trip_id": trip_id, + "arrival_time": s_to_hms(arr_n), + "departure_time": s_to_hms(dep_n), + "arrival_s": arr_n, + "departure_s": dep_n, + "stop_id": new_stop.stop_id, + "stop_sequence": seq_a + 1, + } + ) + + modified_trips.add(trip_id) + + # 1) shift stop_sequence for all following stops + seq_shifts.append((trip_id, seq_a + 1)) + + # 2) shift times according to strategy + if propagate == "after": + time_shifts.append((trip_id, "after", seq_b, delta_total)) + elif propagate == "before": + time_shifts.append((trip_id, "before", seq_a, delta_total)) + else: + half = int(round(delta_total / 2)) + time_shifts.append((trip_id, "before", seq_a, half)) + time_shifts.append((trip_id, "after", seq_b, delta_total - half)) + + # Apply shifts (vectorized masks) + for trip_id, start_seq in seq_shifts: + mask = (st["trip_id"] == trip_id) & (st["stop_sequence"] >= start_seq) + st.loc[mask, "stop_sequence"] = st.loc[mask, "stop_sequence"] + 1 + + for trip_id, mode, boundary_seq, seconds in time_shifts: + if mode == "after": + mask = (st["trip_id"] == trip_id) & (st["stop_sequence"] >= boundary_seq) + st.loc[mask, "arrival_s"] = st.loc[mask, "arrival_s"] + seconds + st.loc[mask, "departure_s"] = st.loc[mask, "departure_s"] + seconds + else: # before + mask = (st["trip_id"] == trip_id) & (st["stop_sequence"] <= boundary_seq) + st.loc[mask, "arrival_s"] = st.loc[mask, "arrival_s"] - seconds + st.loc[mask, "departure_s"] = st.loc[mask, "departure_s"] - seconds + + if insert_rows: + st = pd.concat([st, pd.DataFrame(insert_rows)], ignore_index=True) + st.sort_values(["trip_id", "stop_sequence"], inplace=True) + + # Back to GTFS time strings. + st["arrival_s"] = pd.to_numeric(st["arrival_s"], errors="coerce").round().astype("Int64") + st["departure_s"] = pd.to_numeric(st["departure_s"], errors="coerce").round().astype("Int64") + st["arrival_time"] = st["arrival_s"].map(lambda x: "" if pd.isna(x) else s_to_hms(int(x))) + st["departure_time"] = st["departure_s"].map(lambda x: "" if pd.isna(x) else s_to_hms(int(x))) + st.drop(columns=["arrival_s", "departure_s"], inplace=True) + + feed.tables["stop_times.txt"] = st + + logging.info( + "[GTFS edit] insert_stop_between: done (%s trips modified, %s stop_times rows inserted)", + len(modified_trips), + len(insert_rows), + ) diff --git a/mobility/transport/modes/public_transport/gtfs/gtfs_router.py b/mobility/transport/modes/public_transport/gtfs/gtfs_router.py index 796800ab..e0f95cdc 100644 --- a/mobility/transport/modes/public_transport/gtfs/gtfs_router.py +++ b/mobility/transport/modes/public_transport/gtfs/gtfs_router.py @@ -1,366 +1,321 @@ -import gtfs_kit -import os -import pathlib +""" +GTFS router asset. + +Responsibilities: +- Identify GTFS sources covering the transport zones. +- Optionally apply small "surgical" GTFS edits (e.g. insert a stop between two others). +- Run the R pipeline (prepare_gtfs_router.R) to build a merged Tuesday-only GTFS router (.rds). + +This class is a FileAsset to benefit from caching. +""" + +from __future__ import annotations + import json import logging +import os +import pathlib +from importlib import resources +from typing import Any + +import gtfs_kit import pandas as pd -from importlib import resources from mobility.runtime.assets.file_asset import FileAsset -from mobility.spatial.transport_zones import TransportZones -from mobility.runtime.r_integration.r_script_runner import RScriptRunner - from mobility.runtime.io.download_file import download_file +from mobility.runtime.r_integration.r_script_runner import RScriptRunner +from mobility.spatial.transport_zones import TransportZones from mobility.transport.modes.public_transport.gtfs.gtfs_stops import GTFSStops from .gtfs_data import GTFSData +from .gtfs_edit import apply_gtfs_edits + +LOGGER = logging.getLogger(__name__) + class GTFSRouter(FileAsset): """ - Creates a GTFS router for the given transport zones and saves it in .rds format. - Currently works for France and Switzerland. - - Uses GTFSStops to get a list of the stops within the transport zones, the downloads the GTFS (GTFSData class), - checks that expected agencies are present (if they were provided by the user in PublicTransportRoutingParameters) - and creates the GTFS router using the R script prepare_gtfs_router.R - - For each GTFS source, this script will only keep stops with the region, add missing route types (by default bus), - make IDs unique and remove erroneous calendar dates. - It will then align all GTFS sources on a common start date and merge all GTFS into one. - It adds missing transfers between stops using a crow-fly formula, ans with a limit of 200m. - It finds the Tuesday with the most services running within the montth with the most services on average. - Finally, this global Tuesday-only GTFS is saved. + Create a GTFS router (.rds) for given transport zones. + + It: + - Retrieves GTFS URLs covering the zone bounding box. + - Downloads GTFS files. + - Optionally applies edits (gtfs_edits). + - Optionally checks that expected agencies exist in at least one GTFS. + - Runs prepare_gtfs_router.R to build the router. """ - - def __init__(self, transport_zones: TransportZones, additional_gtfs_files: list = None, expected_agencies: list = None): - + + def __init__( + self, + transport_zones: TransportZones, + additional_gtfs_files: list[str] | None = None, + gtfs_edits: list[dict[str, Any]] | None = None, + expected_agencies: list[str] | None = None, + ): inputs = { "transport_zones": transport_zones, "additional_gtfs_files": additional_gtfs_files, "download_date": os.environ["MOBILITY_GTFS_DOWNLOAD_DATE"], - "expected_agencies": expected_agencies + "expected_agencies": expected_agencies, + "gtfs_edits": gtfs_edits, } - - cache_path = pathlib.Path(os.environ["MOBILITY_PROJECT_DATA_FOLDER"]) / "gtfs_router.rds" + cache_path = pathlib.Path(os.environ["MOBILITY_PROJECT_DATA_FOLDER"]) / "gtfs_router.rds" super().__init__(inputs, cache_path) - + def get_cached_asset(self): return self.cache_path - + def create_and_get_asset(self): - - logging.info("Downloading GTFS files for stops within the transport zones...") - + LOGGER.info("Downloading GTFS files for stops within the transport zones...") + transport_zones = self.inputs["transport_zones"] expected_agencies = self.inputs["expected_agencies"] - + stops = self.get_stops(transport_zones) gtfs_files = self.get_gtfs_files(stops) - - if self.inputs["additional_gtfs_files"] is not None: + if self.inputs["additional_gtfs_files"]: gtfs_files.extend(self.inputs["additional_gtfs_files"]) - - if expected_agencies is not None: + + if self.inputs.get("gtfs_edits"): + # Apply only the lightweight GTFS edits supported by this branch: + # insert a stop between two consecutive stops and propagate the delay. + edits_folder = pathlib.Path(os.environ["MOBILITY_PROJECT_DATA_FOLDER"]) / "gtfs_edits" + gtfs_files = apply_gtfs_edits(gtfs_files, self.inputs["gtfs_edits"], edits_folder) + + if expected_agencies: self.check_expected_agencies(gtfs_files, expected_agencies) - - self.prepare_gtfs_router(transport_zones, gtfs_files) + self.prepare_gtfs_router(transport_zones, gtfs_files) return self.cache_path - - def check_expected_agencies(self, gtfs_files, expected_agencies): - logging.info(gtfs_files) - for gtfs_url in gtfs_files: - logging.info("GTFS") - gtfs=GTFSData(gtfs_url) - agencies = gtfs.get_agencies_names(gtfs_url) - logging.info(agencies) - logging.info(type(agencies)) - for expected_agency in expected_agencies: - logging.info(f'Looking for {expected_agency} in {gtfs.name}') - if expected_agency.lower() in agencies.lower(): - logging.info(f"{expected_agency} found in {gtfs.name}") - expected_agencies.remove(expected_agency) - logging.info(expected_agencies) - if expected_agencies == []: - logging.info("All expected agencies were found") - return True - else: - logging.info("Some agencies were not found in GTFS files.") - logging.info(expected_agencies) - raise IndexError('Missing agencies') - - - def get_stops(self, transport_zones): - - transport_zones = transport_zones.get() - + + def check_expected_agencies(self, gtfs_files: list[str], expected_agencies: list[str]) -> bool: + """ + Ensure each agency in expected_agencies is found in at least one GTFS. + + Note: mutates the expected_agencies list in-place (removes found agencies), + matching the previous behavior. + """ + missing = list(expected_agencies) + + for gtfs_path in gtfs_files: + try: + gtfs = GTFSData(gtfs_path) + agencies = gtfs.get_agencies_names(gtfs_path) + except Exception: + LOGGER.exception("Failed reading agencies for GTFS: %s", gtfs_path) + continue + + for agency in list(missing): + if agency.lower() in str(agencies).lower(): + LOGGER.info("%s found in %s", agency, gtfs.name) + missing.remove(agency) + + if not missing: + LOGGER.info("All expected agencies were found.") + expected_agencies[:] = [] + return True + + LOGGER.error("Some agencies were not found in GTFS files: %s", missing) + raise IndexError("Missing agencies") + + def get_stops(self, transport_zones: TransportZones): + tz = transport_zones.get() + admin_prefixes = ["fr", "ch"] - admin_prefixes = [prefix for prefix in admin_prefixes if transport_zones["local_admin_unit_id"].str.contains(prefix).any()] - + admin_prefixes = [ + prefix for prefix in admin_prefixes if tz["local_admin_unit_id"].str.contains(prefix).any() + ] + stops = GTFSStops(admin_prefixes, self.inputs["download_date"]) - stops = stops.get(bbox=tuple(transport_zones.total_bounds)) - - return stops - - - def prepare_gtfs_router(self, transport_zones, gtfs_files): - - gtfs_files = ",".join(gtfs_files) - - script = RScriptRunner(resources.files('mobility.transport.modes.public_transport.gtfs').joinpath('prepare_gtfs_router.R')) - + return stops.get(bbox=tuple(tz.total_bounds)) + + def prepare_gtfs_router(self, transport_zones: TransportZones, gtfs_files: list[str]) -> None: + gtfs_files_arg = ",".join(gtfs_files) + + script = RScriptRunner( + resources.files("mobility.transport.modes.public_transport.gtfs").joinpath("prepare_gtfs_router.R") + ) script.run( args=[ str(transport_zones.cache_path), - gtfs_files, - str(resources.files('mobility.runtime.resources').joinpath('gtfs/gtfs_route_types.csv')), - str(self.cache_path) + gtfs_files_arg, + str(resources.files("mobility.runtime.resources").joinpath("gtfs/gtfs_route_types.csv")), + str(self.cache_path), ] ) - - return None - - - def get_gtfs_files(self, stops): - + + def get_gtfs_files(self, stops) -> list[str]: gtfs_urls = self.get_gtfs_urls(stops) - gtfs_files = [GTFSData(gtfs_url).get() for gtfs_url in gtfs_urls] - gtfs_files = [str(f[0]) for f in gtfs_files if f[1] == True] - - return gtfs_files - - - - def get_gtfs_urls(self, stops): - - gtfs_urls = [] - - # Add resource urls that are already known (for Switzerland for example) + gtfs_files = [GTFSData(url).get() for url in gtfs_urls] + return [str(f[0]) for f in gtfs_files if f[1] is True] + + def get_gtfs_urls(self, stops) -> list[str]: + gtfs_urls: list[str] = [] + gtfs_urls.extend(stops["resource_url"].dropna().unique().tolist()) - - # Add transport.data.gouv.fr resource urls by matching their datagouv_id in the global metadata file + datagouv_dataset_urls = stops["dataset_url"].dropna().unique() datagouv_dataset_ids = [pathlib.Path(url).name for url in datagouv_dataset_urls] - + url = "https://transport.data.gouv.fr/api/datasets" - path = ( - pathlib.Path(os.environ["MOBILITY_PACKAGE_DATA_FOLDER"]) / "gtfs"/ - (self.inputs["download_date"] + "_gtfs_metadata.json") + path = ( + pathlib.Path(os.environ["MOBILITY_PACKAGE_DATA_FOLDER"]) + / "gtfs" + / (self.inputs["download_date"] + "_gtfs_metadata.json") ) download_file(url, path) - - with open(path, "r", encoding="UTF-8") as f: + + with open(path, "r", encoding="utf-8") as f: metadata = json.load(f) - + for dataset_metadata in metadata: - if dataset_metadata["datagouv_id"] in datagouv_dataset_ids: - gtfs_resources = [r for r in dataset_metadata["resources"] if "format" in r.keys()] - gtfs_resources = [r for r in gtfs_resources if r["format"] == "GTFS"] - for r in gtfs_resources: + if dataset_metadata.get("datagouv_id") not in datagouv_dataset_ids: + continue + + resources_ = dataset_metadata.get("resources", []) + gtfs_resources = [r for r in resources_ if r.get("format") == "GTFS"] + for r in gtfs_resources: + if r.get("original_url"): gtfs_urls.append(r["original_url"]) - + return gtfs_urls - + def audit_gtfs(self): """ - Used to audit and verify GTFS files. - For each GTFS in the defined Transport Zones, the function : - - finds the date with the max number of services - - if file shapes.txt is absent, recreates shapes based on stop sequences - - computes number of trips on each shape for the max services date - - identifies active stops on the max services date - - exports stops and shapes enriched with trip count and route name as a GeoPackage file + Audit GTFS files for the current transport zones. + + Exports (per GTFS source) a GeoPackage with: + - active shapes (busiest date) enriched with trip counts and route names + - active stops (busiest date) """ - transport_zones = self.inputs["transport_zones"] stops = self.get_stops(transport_zones) gtfs_files = self.get_gtfs_files(stops) - for i, gtfs_url in enumerate(gtfs_files, start=1): - logging.info("GTFS") - logging.info(gtfs_url) - gtfs=GTFSData(gtfs_url) - agencies = gtfs.get_agencies_names(gtfs_url) + for i, gtfs_path in enumerate(gtfs_files, start=1): + LOGGER.info("Auditing GTFS: %s", gtfs_path) try: - feed = gtfs_kit.read_feed(gtfs_url, dist_units='m') + feed = gtfs_kit.read_feed(gtfs_path, dist_units="m") except Exception as e: - logging.info(f"Error in loading GTFS : {e}") + LOGGER.info("Error loading GTFS: %s", e) + continue - # 1. Load the reference dataframes (make copies to avoid modifing the feed) - # If shapes.txt et shape_id exist, load them : - try: - shapes_df = feed.shapes.copy() - trips_df = feed.trips[['trip_id', 'route_id', 'shape_id']].copy() - # Otherwise : - except: - shapes_df = None - trips_df = feed.trips[['trip_id', 'route_id']].copy() - - routes_df = feed.routes[['route_id', 'route_short_name', 'route_long_name']].copy() - stops_df = feed.stops[['stop_id','stop_name','stop_lat', 'stop_lon']].copy() + shapes_df, trips_df = self._load_shapes_and_trips(feed) + routes_df = feed.routes[["route_id", "route_short_name", "route_long_name"]].copy() + stops_df = feed.stops[["stop_id", "stop_name", "stop_lat", "stop_lon"]].copy() stop_times_df = feed.stop_times.copy() dates = feed.get_dates() max_services_date = feed.compute_busiest_date(dates) - logging.info(f"Max services date is {max_services_date}") + LOGGER.info("Max services date is %s", max_services_date) - # 2. Filter active trips for the busiest date - #active_trips contains all the trips for the busiest date active_trips = feed.get_trips(date=max_services_date) if active_trips.empty: - logging.info(f"No active trips found for {max_services_date}.") - else : - logging.info(f"{len(active_trips)} trips found for {max_services_date}") - - # 3. Get the active shapes corresponding to the target date, count the trips for each shape and add route name - # 3A. If shapes_df is not null : group active trips by shape_id and count the number of trips for each shape_id - if shapes_df is not None and not shapes_df.empty: - logging.info("File shapes.txt is present in GTFS feed, counting trips...") - - # Group active trips by shape_id and count the number of trips for each shape_id - trips_counts = active_trips.groupby(['shape_id']).size().reset_index(name='trip_count') - - # This part is added to manage the case where the same shapes may have different route_ids - # even though they are on the same route - # Retrieve a unique route_id corresponding to each shape_id - shapes_routes = active_trips.groupby(['shape_id'])['route_id'].first().reset_index() - - # Add route_id to trips_counts - trips_counts = trips_counts.merge( - shapes_routes, - on='shape_id', - how='left' - ) - - # Remove trips with empty shape_id if they are still present - trips_counts = trips_counts[trips_counts['shape_id'].notna()] - - # Create active_shapes_df - active_shapes_df = shapes_df[shapes_df['shape_id'].isin(active_trips['shape_id'])] - - # 3B. If shape_id doesn't exist : reconstruct the shapes - else: - logging.info("File shapes.txt is missing in GTFS feed, reconstructing shapes...") - - # Join active_trips and stop_times_df to have the stop sequence for each trip - trips_stop_sequences = pd.merge( - active_trips[['trip_id','route_id']], - stop_times_df[['trip_id','stop_id','stop_sequence']], - on='trip_id', - how='left' - ) - - # Sort by trip_id and stop_sequence - trips_stop_sequences = trips_stop_sequences.sort_values(by=['trip_id','stop_sequence']) - - # Group by trip_id and concat stops_id in one unique string to rebuild a pseudo_shape_id - trips_with_pseudo_shape_id = trips_stop_sequences.groupby(['trip_id','route_id']).agg( - pseudo_shape_id=('stop_id', lambda x: '-'.join(x.astype(str))) - ).reset_index() - - # Add the pseudo_shape_id to trips_with_stop_sequence - trips_stop_sequences = pd.merge( - trips_stop_sequences, - trips_with_pseudo_shape_id[['trip_id','pseudo_shape_id']], - on='trip_id' - ) - - # Suppress all the non unique values to recreate a shapes df - pseudo_shapes = trips_stop_sequences[[ - 'pseudo_shape_id', - 'stop_id', - 'stop_sequence', - 'route_id' - ]].drop_duplicates(subset=['pseudo_shape_id', 'stop_sequence']) - - # Get the lat and long from stops_df and add stop_coords to pseudo_shapes - stops_coords = stops_df[['stop_id', 'stop_lat', 'stop_lon']].copy() - stops_coords = stops_coords.rename(columns={'stop_lat': 'shape_pt_lat','stop_lon': 'shape_pt_lon'}) - pseudo_shapes = pd.merge( - pseudo_shapes, - stops_coords, - on='stop_id' - ) - - # Finalize the structure (similar to shapes.txt) - # Rename stop_sequence into shape_pt_sequence - active_shapes_df = pseudo_shapes.rename( - columns={'stop_sequence': 'shape_pt_sequence', - 'pseudo_shape_id': 'shape_id'} - ) - # Sort by shape_id and shape_pt_sequence - active_shapes_df = active_shapes_df.sort_values(['shape_id', 'shape_pt_sequence']) - - # Group active trips by shape_id and count the number of trips for each pseudo_shape_id - logging.info("Counting trips...") - trips_counts = trips_with_pseudo_shape_id.groupby(['pseudo_shape_id']).size().reset_index(name='trip_count') - - # This part is added to manage the case where the same pseudo_shapes may have different route_ids - # even though they are on the same route - # Retrieve a unique route_id corresponding to each shape_id - shapes_routes = trips_with_pseudo_shape_id.groupby(['pseudo_shape_id'])['route_id'].first().reset_index() - - # Add route_id to trips_counts - trips_counts = trips_counts.merge( - shapes_routes, - on='pseudo_shape_id', - how='left' - ) - - trips_counts = trips_counts.rename(columns={'pseudo_shape_id': 'shape_id'}) - - # Add route names to trips_counts - trips_counts = trips_counts.merge( - routes_df, - on='route_id', - how='left' - ) - - # 4. Get the active stops corresponding to the target date - # Keep only stop times for active trips - active_stop_times = stop_times_df[stop_times_df['trip_id'].isin(active_trips['trip_id'])] - - # Extract the corresponding active stop_ids - active_stop_ids = active_stop_times['stop_id'].unique() - - # Filter stops_df to only keep active stops - active_stops_df = stops_df[stops_df['stop_id'].isin(active_stop_ids)] - - # 5. Enrich active_shapes_df and export gpkg - # Creating GeoDataFrames for shapes and stops - active_shapes_gdf = gtfs_kit.shapes.geometrize_shapes(active_shapes_df) - active_stops_gdf = gtfs_kit.stops.geometrize_stops(active_stops_df) - - # Enrich shapes with trip_counts, route_id and route_names - logging.info('Enriching shapes with trip counts and route names...') - active_shapes_gdf = active_shapes_gdf.merge( - trips_counts, - on='shape_id', - how='left' - ) - - # Print some data about number of trips - nb_shapes = active_shapes_gdf['shape_id'].nunique() - trips_total = active_shapes_gdf['trip_count'].sum() - logging.info(f"The network has {nb_shapes} different shapes with a total of {trips_total} trips on {max_services_date}") - - # Replace NaN values of trip_count by 0 - active_shapes_gdf['trip_count'] = active_shapes_gdf['trip_count'].fillna(0) - #active_shapes_gdf = active_shapes_gdf[active_shapes_gdf['trip_count'] > 0.0] - - # Export shapes and stops to gpkg - output_path = pathlib.Path(os.environ["MOBILITY_PACKAGE_DATA_FOLDER"]) / "gtfs" / "gpkg" / f"gtfs_{i}.gpkg" - output_path.parent.mkdir(parents=True, exist_ok=True) - - active_shapes_gdf.to_file(output_path,driver="GPKG",layer="shapes") - active_stops_gdf.to_file(output_path,driver="GPKG",layer="stops") - - # Print some info - logging.info(f"GTFS stops and shapes exported as GeoPackage in file {output_path}") - - - - + LOGGER.info("No active trips found for %s.", max_services_date) + continue + + LOGGER.info("%s trips found for %s", len(active_trips), max_services_date) + + active_shapes_df, trips_counts = self._build_active_shapes( + active_trips=active_trips, + shapes_df=shapes_df, + stop_times_df=stop_times_df, + stops_df=stops_df, + ) + + trips_counts = trips_counts.merge(routes_df, on="route_id", how="left") + + active_stop_times = stop_times_df[stop_times_df["trip_id"].isin(active_trips["trip_id"])] + active_stop_ids = active_stop_times["stop_id"].unique() + active_stops_df = stops_df[stops_df["stop_id"].isin(active_stop_ids)] + + active_shapes_gdf = gtfs_kit.shapes.geometrize_shapes(active_shapes_df) + active_stops_gdf = gtfs_kit.stops.geometrize_stops(active_stops_df) + + LOGGER.info("Enriching shapes with trip counts and route names...") + active_shapes_gdf = active_shapes_gdf.merge(trips_counts, on="shape_id", how="left") + + nb_shapes = active_shapes_gdf["shape_id"].nunique() + trips_total = active_shapes_gdf["trip_count"].sum() + LOGGER.info( + "Network has %s shapes with a total of %s trips on %s", + nb_shapes, + trips_total, + max_services_date, + ) + + active_shapes_gdf["trip_count"] = active_shapes_gdf["trip_count"].fillna(0) + + output_path = ( + pathlib.Path(os.environ["MOBILITY_PACKAGE_DATA_FOLDER"]) / "gtfs" / "gpkg" / f"gtfs_{i}.gpkg" + ) + output_path.parent.mkdir(parents=True, exist_ok=True) + + active_shapes_gdf.to_file(output_path, driver="GPKG", layer="shapes") + active_stops_gdf.to_file(output_path, driver="GPKG", layer="stops") + + LOGGER.info("GTFS stops and shapes exported as GeoPackage in %s", output_path) + + @staticmethod + def _load_shapes_and_trips(feed): + try: + shapes_df = feed.shapes.copy() + trips_df = feed.trips[["trip_id", "route_id", "shape_id"]].copy() + except Exception: + shapes_df = None + trips_df = feed.trips[["trip_id", "route_id"]].copy() + return shapes_df, trips_df + + @staticmethod + def _build_active_shapes(active_trips, shapes_df, stop_times_df, stops_df): + if shapes_df is not None and not shapes_df.empty: + LOGGER.info("shapes.txt present: counting trips by shape_id...") + trips_counts = active_trips.groupby(["shape_id"]).size().reset_index(name="trip_count") + shapes_routes = active_trips.groupby(["shape_id"])["route_id"].first().reset_index() + trips_counts = trips_counts.merge(shapes_routes, on="shape_id", how="left") + trips_counts = trips_counts[trips_counts["shape_id"].notna()] + active_shapes_df = shapes_df[shapes_df["shape_id"].isin(active_trips["shape_id"])] + return active_shapes_df, trips_counts + + LOGGER.info("shapes.txt missing: reconstructing shapes from stop sequences...") + trips_stop_sequences = pd.merge( + active_trips[["trip_id", "route_id"]], + stop_times_df[["trip_id", "stop_id", "stop_sequence"]], + on="trip_id", + how="left", + ).sort_values(by=["trip_id", "stop_sequence"]) + + trips_with_pseudo_shape_id = trips_stop_sequences.groupby(["trip_id", "route_id"]).agg( + pseudo_shape_id=("stop_id", lambda x: "-".join(x.astype(str))) + ).reset_index() + + trips_stop_sequences = pd.merge( + trips_stop_sequences, + trips_with_pseudo_shape_id[["trip_id", "pseudo_shape_id"]], + on="trip_id", + ) + + pseudo_shapes = trips_stop_sequences[ + ["pseudo_shape_id", "stop_id", "stop_sequence", "route_id"] + ].drop_duplicates(subset=["pseudo_shape_id", "stop_sequence"]) + + stops_coords = stops_df[["stop_id", "stop_lat", "stop_lon"]].copy().rename( + columns={"stop_lat": "shape_pt_lat", "stop_lon": "shape_pt_lon"} + ) + pseudo_shapes = pd.merge(pseudo_shapes, stops_coords, on="stop_id") + + active_shapes_df = pseudo_shapes.rename( + columns={"stop_sequence": "shape_pt_sequence", "pseudo_shape_id": "shape_id"} + ).sort_values(["shape_id", "shape_pt_sequence"]) + + LOGGER.info("Counting trips by reconstructed shape_id...") + trips_counts = trips_with_pseudo_shape_id.groupby(["pseudo_shape_id"]).size().reset_index(name="trip_count") + shapes_routes = trips_with_pseudo_shape_id.groupby(["pseudo_shape_id"])["route_id"].first().reset_index() + trips_counts = trips_counts.merge(shapes_routes, on="pseudo_shape_id", how="left").rename( + columns={"pseudo_shape_id": "shape_id"} + ) + + return active_shapes_df, trips_counts diff --git a/mobility/transport/modes/public_transport/public_transport_graph.py b/mobility/transport/modes/public_transport/public_transport_graph.py index bf3c4043..dafd1ec2 100644 --- a/mobility/transport/modes/public_transport/public_transport_graph.py +++ b/mobility/transport/modes/public_transport/public_transport_graph.py @@ -5,7 +5,7 @@ import pandas as pd import geopandas as gpd import numpy as np -from typing import Annotated +from typing import Annotated, Any from importlib import resources from pydantic import BaseModel, ConfigDict, Field, model_validator @@ -46,9 +46,10 @@ def __init__( parameters = PublicTransportRoutingParameters() gtfs_router = GTFSRouter( - transport_zones, - parameters.additional_gtfs_files, - parameters.expected_agencies + transport_zones=transport_zones, + additional_gtfs_files=parameters.additional_gtfs_files, + gtfs_edits=parameters.gtfs_edits, + expected_agencies=parameters.expected_agencies, ) inputs = { @@ -121,6 +122,7 @@ class PublicTransportRoutingParameters(BaseModel): These parameters combine: - a coarse outer OD envelope through `max_beeline_distance`, in km - time-window and generalized-time constraints for the public transport leg + - optional GTFS source selection and edit rules `max_beeline_distance` is only used to prune obviously too-distant OD pairs before detailed multimodal routing. It does not replace the detailed public @@ -143,6 +145,7 @@ class PublicTransportRoutingParameters(BaseModel): Field(default=DEFAULT_LONG_RANGE_MOTORIZED_MAX_BEELINE_DISTANCE_KM, gt=0.0), ] additional_gtfs_files: Annotated[ListParameterProfile | list[str] | None, Field(default=None)] + gtfs_edits: Annotated[ListParameterProfile | list[dict[str, Any]] | None, Field(default=None)] expected_agencies: Annotated[list[str] | None, Field(default=None)] @model_validator(mode="after") diff --git a/tests/back/unit/costs/travel_costs/test_001_routing_parameters.py b/tests/back/unit/costs/travel_costs/test_001_routing_parameters.py index a942c7b7..cf138b5d 100644 --- a/tests/back/unit/costs/travel_costs/test_001_routing_parameters.py +++ b/tests/back/unit/costs/travel_costs/test_001_routing_parameters.py @@ -53,3 +53,28 @@ def test_public_transport_routing_parameters_resolve_list_profiles_by_iteration( assert step_1.additional_gtfs_files == ["base.zip"] assert step_2.additional_gtfs_files == ["base.zip", "event.zip"] + + +def test_public_transport_routing_parameters_accept_gtfs_edits(): + params = PublicTransportRoutingParameters( + gtfs_edits=[{"mode": "all", "ops": []}], + ) + + assert params.gtfs_edits == [{"mode": "all", "ops": []}] + + +def test_public_transport_routing_parameters_resolve_gtfs_edits_profiles_by_iteration(): + params = PublicTransportRoutingParameters( + gtfs_edits=ListParameterProfile( + points={ + 1: [{"mode": "all", "ops": []}], + 2: [{"mode": "all", "ops": [{"action": "add_stop"}]}], + } + ) + ) + + step_1 = resolve_model_for_iteration(params, 1) + step_2 = resolve_model_for_iteration(params, 2) + + assert step_1.gtfs_edits == [{"mode": "all", "ops": []}] + assert step_2.gtfs_edits == [{"mode": "all", "ops": [{"action": "add_stop"}]}] diff --git a/tests/back/unit/transport/modes/public_transport/gtfs/test_gtfs_edit.py b/tests/back/unit/transport/modes/public_transport/gtfs/test_gtfs_edit.py new file mode 100644 index 00000000..2cf6e472 --- /dev/null +++ b/tests/back/unit/transport/modes/public_transport/gtfs/test_gtfs_edit.py @@ -0,0 +1,229 @@ +from pathlib import Path + +import pytest + +from mobility.transport.modes.public_transport.gtfs.gtfs_edit import ( + GTFSFeed, + NewStop, + apply_gtfs_edits, + insert_stop_between, +) +from mobility.transport.modes.public_transport.gtfs_builder import ( + GTFSFeedSpec, + GTFSLineSpec, + GTFSStopSpec, + build_gtfs_zip, +) + + +def _build_simple_gtfs_zip( + tmp_path: Path, + *, + bidirectional: bool, + start_time: float = 0.0, + end_time: float = 0.0, +) -> Path: + feed = GTFSFeedSpec( + agency_id="agency_1", + agency_name="Test agency", + route_id="route_1", + route_short_name="T1", + route_type="bus", + service_id="service_1", + stops={ + "A": GTFSStopSpec(lon=6.0, lat=46.0, name="Stop A"), + "B": GTFSStopSpec(lon=6.01, lat=46.01, name="Stop B"), + "C": GTFSStopSpec(lon=6.02, lat=46.02, name="Stop C"), + }, + lines=[ + GTFSLineSpec( + stop_ids=["A", "B", "C"], + segment_travel_times=[600.0, 600.0], + start_time=start_time, + end_time=end_time, + period=1.0, + bidirectional=bidirectional, + ) + ], + ) + + output_path = tmp_path / "simple_feed.zip" + return build_gtfs_zip(feed, output_path) + + +def test_gtfs_feed_roundtrip_and_insert_stop_between(tmp_path): + gtfs_path = _build_simple_gtfs_zip(tmp_path, bidirectional=False) + + feed = GTFSFeed(gtfs_path).load() + roundtrip_path = tmp_path / "roundtrip.zip" + feed.save(roundtrip_path) + + roundtrip = GTFSFeed(roundtrip_path).load() + assert set(roundtrip.tables) == { + "agency.txt", + "routes.txt", + "trips.txt", + "calendar.txt", + "stops.txt", + "stop_times.txt", + } + + insert_stop_between( + feed=feed, + from_stop_id="A", + to_stop_id="B", + new_stop=NewStop( + stop_id="X", + stop_name="Inserted stop", + stop_lat=46.005, + stop_lon=6.005, + ), + dwell_time_s=45, + extra_run_time_s=30, + split_ratio=0.5, + propagate="after", + ) + + stops = feed.tables["stops.txt"] + stop_times = feed.tables["stop_times.txt"].sort_values(["trip_id", "stop_sequence"]).reset_index(drop=True) + + assert "X" in stops["stop_id"].tolist() + assert stop_times["stop_id"].tolist() == ["A", "X", "B", "C"] + assert stop_times["stop_sequence"].tolist() == [1, 2, 3, 4] + assert stop_times.loc[1, "arrival_time"] == "00:05:00" + assert stop_times.loc[1, "departure_time"] == "00:05:45" + assert stop_times.loc[2, "arrival_time"] == "00:11:15" + assert stop_times.loc[3, "arrival_time"] == "00:21:15" + + +def test_apply_gtfs_edits_creates_and_reuses_edited_feed(tmp_path): + gtfs_path = _build_simple_gtfs_zip(tmp_path, bidirectional=True) + edits_folder = tmp_path / "edits" + + edits = [ + { + "path": str(gtfs_path), + "mode": "explicit", + "ops": [ + { + "op": "insert_stop_between", + "from_stop_id": "A", + "to_stop_id": "B", + "new_stop": { + "stop_id": "X", + "stop_name": "Inserted stop", + "stop_lat": 46.005, + "stop_lon": 6.005, + }, + "dwell_time_s": 45, + "extra_run_time_s": 30, + "split_ratio": 0.5, + "propagate": "after", + "bidirectional": True, + } + ], + } + ] + + edited_files = apply_gtfs_edits([str(gtfs_path)], edits, edits_folder) + assert len(edited_files) == 1 + edited_path = Path(edited_files[0]) + assert edited_path.exists() + + edited_feed = GTFSFeed(edited_path).load() + stop_times = edited_feed.tables["stop_times.txt"] + + assert stop_times.shape[0] == 8 + assert stop_times["stop_id"].tolist().count("X") == 2 + + reused_files = apply_gtfs_edits([str(gtfs_path)], edits, edits_folder) + assert reused_files == edited_files + + +def test_apply_gtfs_edits_mode_all_uses_chain_matches(tmp_path): + gtfs_path = _build_simple_gtfs_zip(tmp_path, bidirectional=True) + edits_folder = tmp_path / "all_edits" + + edits = [ + { + "mode": "all", + "ops": [ + { + "op": "insert_stop_between", + "from_stop_id": "A", + "to_stop_id": "B", + "new_stop": { + "stop_id": "X", + "stop_name": "Inserted stop", + "stop_lat": 46.005, + "stop_lon": 6.005, + }, + "dwell_time_s": 45, + "extra_run_time_s": 30, + "split_ratio": 0.5, + "propagate": "after", + "bidirectional": True, + } + ], + } + ] + + edited_files = apply_gtfs_edits([str(gtfs_path)], edits, edits_folder) + assert len(edited_files) == 1 + + edited_feed = GTFSFeed(edited_files[0]).load() + stop_times = edited_feed.tables["stop_times.txt"] + + assert stop_times.shape[0] == 8 + assert stop_times["stop_id"].tolist().count("X") == 2 + + +@pytest.mark.parametrize( + ("propagate", "expected_a_time", "expected_b_time", "expected_c_time"), + [ + ("before", "00:58:45", "01:10:00", "01:20:00"), + ("symmetric", "00:59:22", "01:10:37", "01:20:37"), + ], +) +def test_insert_stop_between_supports_before_and_symmetric( + tmp_path, + propagate, + expected_a_time, + expected_b_time, + expected_c_time, +): + gtfs_path = _build_simple_gtfs_zip( + tmp_path, + bidirectional=False, + start_time=3600.0, + end_time=3600.0, + ) + + feed = GTFSFeed(gtfs_path).load() + insert_stop_between( + feed=feed, + from_stop_id="A", + to_stop_id="B", + new_stop=NewStop( + stop_id="X", + stop_name="Inserted stop", + stop_lat=46.005, + stop_lon=6.005, + ), + dwell_time_s=45, + extra_run_time_s=30, + split_ratio=0.5, + propagate=propagate, + ) + + stop_times = feed.tables["stop_times.txt"].sort_values(["trip_id", "stop_sequence"]).reset_index(drop=True) + + assert stop_times["stop_id"].tolist() == ["A", "X", "B", "C"] + assert stop_times.loc[0, "arrival_time"] == expected_a_time + assert stop_times.loc[0, "departure_time"] == expected_a_time + assert stop_times.loc[1, "arrival_time"] == "01:05:00" + assert stop_times.loc[1, "departure_time"] == "01:05:45" + assert stop_times.loc[2, "arrival_time"] == expected_b_time + assert stop_times.loc[2, "departure_time"] == expected_b_time + assert stop_times.loc[3, "arrival_time"] == expected_c_time + assert stop_times.loc[3, "departure_time"] == expected_c_time diff --git a/tests/back/unit/transport/modes/public_transport/gtfs/test_gtfs_router.py b/tests/back/unit/transport/modes/public_transport/gtfs/test_gtfs_router.py new file mode 100644 index 00000000..59744808 --- /dev/null +++ b/tests/back/unit/transport/modes/public_transport/gtfs/test_gtfs_router.py @@ -0,0 +1,278 @@ +from __future__ import annotations + +import json +from pathlib import Path +from types import SimpleNamespace + +import geopandas as gpd +import pandas as pd +import pytest + +from mobility.transport.modes.public_transport.gtfs import gtfs_router as gtfs_router_module +from mobility.transport.modes.public_transport.gtfs.gtfs_router import GTFSRouter + + +def _set_router_env(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> tuple[Path, Path]: + project_data_folder = tmp_path / "project-data" + package_data_folder = tmp_path / "package-data" + project_data_folder.mkdir(parents=True, exist_ok=True) + package_data_folder.mkdir(parents=True, exist_ok=True) + monkeypatch.setenv("MOBILITY_GTFS_DOWNLOAD_DATE", "2024-01-01") + monkeypatch.setenv("MOBILITY_PROJECT_DATA_FOLDER", str(project_data_folder)) + monkeypatch.setenv("MOBILITY_PACKAGE_DATA_FOLDER", str(package_data_folder)) + return project_data_folder, package_data_folder + + +def test_create_and_get_asset_applies_edits_checks_expected_agencies_and_prepares_router( + monkeypatch, + tmp_path, +): + project_data_folder, _ = _set_router_env(monkeypatch, tmp_path) + + calls = {} + + monkeypatch.setattr( + GTFSRouter, + "get_stops", + lambda self, transport_zones: pd.DataFrame( + {"resource_url": ["/tmp/base.zip"], "dataset_url": [None]} + ), + ) + monkeypatch.setattr(GTFSRouter, "get_gtfs_files", lambda self, stops: ["/tmp/base.zip"]) + + def fake_apply_gtfs_edits(gtfs_files, gtfs_edits, edits_folder): + calls["apply_gtfs_edits"] = (list(gtfs_files), gtfs_edits, Path(edits_folder)) + return ["/tmp/edited.zip"] + + def fake_check_expected_agencies(self, gtfs_files, expected_agencies): + calls["expected_agencies"] = (list(gtfs_files), list(expected_agencies)) + expected_agencies[:] = [] + return True + + def fake_prepare_gtfs_router(self, transport_zones, gtfs_files): + calls["prepare_gtfs_router"] = (transport_zones, list(gtfs_files)) + + monkeypatch.setattr(gtfs_router_module, "apply_gtfs_edits", fake_apply_gtfs_edits) + monkeypatch.setattr(GTFSRouter, "check_expected_agencies", fake_check_expected_agencies) + monkeypatch.setattr(GTFSRouter, "prepare_gtfs_router", fake_prepare_gtfs_router) + + router = GTFSRouter( + transport_zones="dummy-transport-zones", + additional_gtfs_files=["/tmp/additional.zip"], + gtfs_edits=[{"mode": "all", "ops": []}], + expected_agencies=["SNCF"], + ) + + result = router.create_and_get_asset() + + assert result == router.cache_path + assert calls["apply_gtfs_edits"][0] == ["/tmp/base.zip", "/tmp/additional.zip"] + assert calls["apply_gtfs_edits"][2] == project_data_folder / "gtfs_edits" + assert calls["expected_agencies"][0] == ["/tmp/edited.zip"] + assert calls["prepare_gtfs_router"][0] == "dummy-transport-zones" + assert calls["prepare_gtfs_router"][1] == ["/tmp/edited.zip"] + + +def test_check_expected_agencies_mutates_expected_list_and_raises_when_missing( + monkeypatch, + tmp_path, +): + _set_router_env(monkeypatch, tmp_path) + + class FakeGTFSData: + def __init__(self, url): + self.url = url + self.name = f"fake-{Path(url).stem}" + + def get_agencies_names(self, gtfs_path): + if "broken" in self.url: + raise RuntimeError("boom") + return "SNCF, Keolis" + + monkeypatch.setattr(gtfs_router_module, "GTFSData", FakeGTFSData) + + router = GTFSRouter(transport_zones="dummy") + + expected_agencies = ["sncf", "keolis"] + assert router.check_expected_agencies(["ok.zip"], expected_agencies) is True + assert expected_agencies == [] + + with pytest.raises(IndexError): + router.check_expected_agencies(["broken.zip"], ["missing"]) + + +def test_get_stops_and_get_gtfs_urls_collect_expected_sources(monkeypatch, tmp_path): + _, package_data_folder = _set_router_env(monkeypatch, tmp_path) + + recorded = {} + + class FakeGTFSStops: + def __init__(self, admin_prefixes, download_date): + recorded["admin_prefixes"] = list(admin_prefixes) + recorded["download_date"] = download_date + + def get(self, bbox): + recorded["bbox"] = bbox + return pd.DataFrame( + { + "resource_url": ["https://example.com/direct.zip"], + "dataset_url": ["https://data.gouv.fr/datasets/dataset-1"], + } + ) + + monkeypatch.setattr(gtfs_router_module, "GTFSStops", FakeGTFSStops) + monkeypatch.setattr(gtfs_router_module, "download_file", lambda url, path: None) + + transport_zones = SimpleNamespace( + cache_path=tmp_path / "transport-zones.rds", + get=lambda: gpd.GeoDataFrame( + {"local_admin_unit_id": ["fr-1", "ch-2"]}, + geometry=gpd.points_from_xy([0.0, 1.0], [0.0, 1.0]), + crs=4326, + ), + ) + + router = GTFSRouter(transport_zones="dummy") + + stops = router.get_stops(transport_zones) + assert recorded["admin_prefixes"] == ["fr", "ch"] + assert recorded["download_date"] == "2024-01-01" + assert "resource_url" in stops.columns + + metadata_path = package_data_folder / "gtfs" / "2024-01-01_gtfs_metadata.json" + metadata_path.parent.mkdir(parents=True, exist_ok=True) + metadata_path.write_text( + json.dumps( + [ + { + "datagouv_id": "dataset-1", + "resources": [ + { + "format": "GTFS", + "original_url": "https://example.com/metadata.zip", + } + ], + } + ] + ), + encoding="utf-8", + ) + + urls = router.get_gtfs_urls(stops) + assert "https://example.com/direct.zip" in urls + assert "https://example.com/metadata.zip" in urls + + +def test_prepare_gtfs_router_uses_rscript_and_route_types_resource(monkeypatch, tmp_path): + _set_router_env(monkeypatch, tmp_path) + + calls = {} + + class FakeRScriptRunner: + def __init__(self, script_path): + calls["script_path"] = Path(script_path) + + def run(self, args): + calls["args"] = list(args) + + monkeypatch.setattr(gtfs_router_module, "RScriptRunner", FakeRScriptRunner) + + router = GTFSRouter(transport_zones="dummy") + transport_zones = SimpleNamespace(cache_path=tmp_path / "transport-zones.rds") + + router.prepare_gtfs_router(transport_zones, ["/tmp/a.zip", "/tmp/b.zip"]) + + assert calls["script_path"].name == "prepare_gtfs_router.R" + assert calls["args"][0] == str(transport_zones.cache_path) + assert calls["args"][1] == "/tmp/a.zip,/tmp/b.zip" + assert calls["args"][2].endswith("gtfs/gtfs_route_types.csv") + assert calls["args"][3] == str(router.cache_path) + + +def test_audit_gtfs_exports_active_shapes_and_stops(monkeypatch, tmp_path): + _set_router_env(monkeypatch, tmp_path) + + class FakeFeed: + def __init__(self): + self.shapes = pd.DataFrame( + { + "shape_id": ["shape-1", "shape-1"], + "shape_pt_lat": [46.0, 46.1], + "shape_pt_lon": [6.0, 6.1], + "shape_pt_sequence": [1, 2], + } + ) + self.trips = pd.DataFrame( + { + "trip_id": ["trip-1"], + "route_id": ["route-1"], + "shape_id": ["shape-1"], + } + ) + self.routes = pd.DataFrame( + { + "route_id": ["route-1"], + "route_short_name": ["R1"], + "route_long_name": ["Route 1"], + } + ) + self.stops = pd.DataFrame( + { + "stop_id": ["A", "B"], + "stop_name": ["Stop A", "Stop B"], + "stop_lat": [46.0, 46.1], + "stop_lon": [6.0, 6.1], + } + ) + self.stop_times = pd.DataFrame( + { + "trip_id": ["trip-1", "trip-1"], + "stop_id": ["A", "B"], + "stop_sequence": [1, 2], + } + ) + + def get_dates(self): + return ["20240101"] + + def compute_busiest_date(self, dates): + return dates[0] + + def get_trips(self, date): + return self.trips.copy() + + recorded = {"to_file": []} + + def fake_geometrize_shapes(df): + return gpd.GeoDataFrame( + df.copy(), + geometry=gpd.points_from_xy(df["shape_pt_lon"], df["shape_pt_lat"]), + crs=4326, + ) + + def fake_geometrize_stops(df): + return gpd.GeoDataFrame( + df.copy(), + geometry=gpd.points_from_xy(df["stop_lon"], df["stop_lat"]), + crs=4326, + ) + + def fake_to_file(self, output_path, driver=None, layer=None): + recorded["to_file"].append((Path(output_path), driver, layer)) + + monkeypatch.setattr(gtfs_router_module.gtfs_kit, "read_feed", lambda gtfs_path, dist_units="m": FakeFeed()) + monkeypatch.setattr(gtfs_router_module.gtfs_kit.shapes, "geometrize_shapes", fake_geometrize_shapes) + monkeypatch.setattr(gtfs_router_module.gtfs_kit.stops, "geometrize_stops", fake_geometrize_stops) + monkeypatch.setattr(gpd.GeoDataFrame, "to_file", fake_to_file, raising=False) + + monkeypatch.setattr(GTFSRouter, "get_stops", lambda self, transport_zones: pd.DataFrame({"resource_url": []})) + monkeypatch.setattr(GTFSRouter, "get_gtfs_files", lambda self, stops: [str(tmp_path / "fake.zip")]) + + router = GTFSRouter(transport_zones="dummy") + router.audit_gtfs() + + assert len(recorded["to_file"]) == 2 + assert recorded["to_file"][0][1:] == ("GPKG", "shapes") + assert recorded["to_file"][1][1:] == ("GPKG", "stops") + assert recorded["to_file"][0][0].name == "gtfs_1.gpkg" + assert recorded["to_file"][1][0].name == "gtfs_1.gpkg" diff --git a/tests/back/unit/transport/modes/public_transport/test_public_transport_graph.py b/tests/back/unit/transport/modes/public_transport/test_public_transport_graph.py new file mode 100644 index 00000000..1ab5c735 --- /dev/null +++ b/tests/back/unit/transport/modes/public_transport/test_public_transport_graph.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +from pathlib import Path + +from mobility.transport.modes.public_transport import public_transport_graph as pt_graph_module +from mobility.transport.modes.public_transport.public_transport_graph import ( + PublicTransportGraph, + PublicTransportRoutingParameters, +) + + +def test_public_transport_graph_forwards_gtfs_edits_to_router(monkeypatch, tmp_path): + monkeypatch.setenv("MOBILITY_PROJECT_DATA_FOLDER", str(tmp_path)) + + calls = {} + + def fake_gtfs_router(transport_zones, additional_gtfs_files, gtfs_edits, expected_agencies): + calls["gtfs_router"] = ( + transport_zones, + additional_gtfs_files, + gtfs_edits, + expected_agencies, + ) + return "fake-gtfs-router" + + monkeypatch.setattr(pt_graph_module, "GTFSRouter", fake_gtfs_router) + + params = PublicTransportRoutingParameters( + additional_gtfs_files=["base.zip"], + gtfs_edits=[{"mode": "all", "ops": []}], + expected_agencies=["SNCF"], + ) + + graph = PublicTransportGraph("dummy-transport-zones", params) + + assert graph.cache_path.parent == Path(tmp_path) / "public_transport_graph" / "simplified" + assert graph.cache_path.name.endswith("-public-transport-graph") + assert calls["gtfs_router"] == ( + "dummy-transport-zones", + ["base.zip"], + [{"mode": "all", "ops": []}], + ["SNCF"], + )