laurel.pipelines.compute_routes package

Submodules

laurel.pipelines.compute_routes.nodes module

Kedro pipeline nodes for the compute_routes pipeline (Model Module 2 — Optional truck-stop dwells).

Inserts optional dwell events at public truck-stop locations that lie along the shortest-path route between each trip’s origin and destination. This implements the first part of Model Module 2 (Augment Dwell Data): if a vehicle travels through a truck stop, it could have stopped there to charge, even if no dwell was recorded in the telematics data. The pipeline uses a self-hosted GraphHopper routing engine to compute shortest-path routes, then performs a spatial join to find truck stops within a buffer of each route.

Pipeline overview

1. import_graph — Imports an OSM road network into a GraphHopper Docker container to prepare it for routing queries.

2. test_get_routes — Sends a single cross-country test query to verify that the GraphHopper server is healthy before batch routing begins.

3. select_trips_to_route — Drops unnecessary columns and splits trips into those worth routing and those below min_dist_miles; optionally subsamples the to-route side for debugging.

4. index_by_vehicle — Indexes a trips DataFrame by vehicle ID with known divisions. Run once, on the full trips set before select_trips_to_route, so both the to-route and not-to-route sides inherit it for free, and it survives unchanged all the way through describe_optional_stop_trips.

5. get_trip_orig_dest_points — Converts origin and destination H3 hexagons to point geometries and attaches them to the trips GeoDataFrame.

6. partition_trips — Re-partitions the Dask GeoDataFrame to the desired number of partitions before routing (allows checkpointing to disk).

7. get_routes_node — Calls get_routes partition-by-partition via GraphHopper, converting raw metric distances and seconds to miles and hours, and setting the route LineString as the active geometry.

8. format_stop_locations — Reformats and point-geometrises the truck-stop candidate locations (Jason’s Law + OSM) for spatial joining.

9. get_optional_stop_trips — Per partition, spatially joins truck stops within a buffer of each route, projects each stop onto the route line to obtain its distance from the trip origin, drops optional stops too close to either trip endpoint, and returns a lazy Dask DataFrame combining original trips with the surviving optional intermediate trips.

10. describe_optional_stop_trips — Per partition, sorts by trip and distance and recomputes start/end timestamps for each sub-segment using proportional time allocation.

11. concat_optional_stops — Combines the never-routed trips with the described optional-stop trips via a divisions-aware dd.concat, then sorts the result by trip ID.

Key design decisions

• GraphHopper containerised routing: Running GraphHopper in a Docker container on the same machine avoids network latency and rate limits associated with hosted routing APIs, which is critical given the ~20 million origin-destination pairs in the full dataset.

• Proportional time allocation: When a trip is split at an optional stop, the time for each sub-segment is computed as (segment_miles / route_speed) × (observed_hours / route_hours) to preserve the observed start and end timestamps while distributing time proportionally to distance.

• Endpoint buffer exclusion: Optional stops within park_buffer_miles of the trip origin or destination are dropped, as the vehicle would most likely have been counted as dwelling there already.

• Deduplication by construction, not anti-join: trips too short to route and routed trips are disjoint sets from the moment select_trips_to_route splits them – a trip can only ever appear in one of the two collections concat_optional_stops combines, so no anti-join is needed to avoid double-counting split trips.

• Divisions survive the checkpoint, not a second shuffle: the vehicle-ID index and its divisions are established once (index_by_vehicle, before select_trips_to_route) and never shuffled again – every node between there and describe_optional_stop_trips operates per-partition only (see _build_partition_trips’s no-shuffle invariant). The catalog entry describe_optional_stop_trips writes to is loaded back with calculate_divisions: True, which recovers known divisions from the written partitions’ index ranges instead of paying for an unsorted set_index (divisions-sampling pass, then data-transfer pass) a second time.

References

Passow, F., & Rajagopal, R. (2026). Identifying indicators to inform proactive substation upgrades for charging electric heavy-duty trucks. Applied Energy.

GraphHopper. Open Source Routing Engine. https://www.graphhopper.com/ U.S. DOT Federal Highway Administration. Jason’s Law Truck Parking Survey.

laurel.pipelines.compute_routes.nodes.concat_optional_stops(trips_not_to_route, trips_opt, params)

Merge optional-stop sub-trips with the trips that were never routed.

trips_not_to_route (trips too short to route – see select_trips_to_route) and trips_opt (output of describe_optional_stop_trips) are already disjoint by construction: a trip is either too short to route, or it went through routing and appears exactly once in trips_opt, split or not. No anti-join/dedup is needed. Both inputs also arrive already indexed by vehicle ID, each with its own known divisions, so they can be combined via interleave_partitions=True – a partition-wise merge on those divisions rather than a full hash shuffle.

A final per-partition stable sort handles the remaining trip-ID column(s) within each vehicle’s rows, since interleaving doesn’t guarantee that ordering on its own.

Parameters:

• trips_not_to_route (DataFrame) – Trips too short to route, indexed by vehicle ID (output of select_trips_to_route, which inherits the index set by index_by_vehicle on the full trips set upstream).

• trips_opt (DataFrame) – Optional-stop sub-trips, indexed by vehicle ID (output of describe_optional_stop_trips, checkpointed to disk with divisions recovered on load – see that function’s docstring).

• params (dict) –

  Pipeline parameters dict with keys:

  • trip_id_cols (list[str]): columns uniquely identifying a trip row; the leading column is the shared index; any remaining columns are sorted on per-partition below.

  • n_partitions (int): target partition count for the merged output. interleave_partitions=True builds output partitions from the union of both inputs’ division boundaries, which can far exceed either input’s own partition count; this rebalances it back down. Divisions stay known and monotonic going in, so this is a cheap merge of adjacent partitions, not a shuffle.

Return type:

DataFrame

Returns:

A Dask DataFrame combining the never-routed trips and the described optional-stop trips, with one row per unique (trip_id_cols) combination, indexed by the leading trip_id_cols entry.

laurel.pipelines.compute_routes.nodes.describe_optional_stop_trips(trips, params)

Compute split timestamps and distances for optional-stop sub-trips.

Applies _describe_partition lazily to each partition of trips (output of get_optional_stop_trips). Safe to run per-partition because every trip’s rows are guaranteed to live in one partition (see _build_partition_trips’s docstring).

The output stays indexed by vehicle ID, same as the input – no shuffle happens here (see _build_partition_trips’s no-shuffle invariant), so the existing divisions remain structurally valid. The catalog entry this feeds is checkpointed to disk with calculate_divisions: True on load, which recovers known divisions from the written partitions’ index ranges for free – no second index_by_vehicle/set_index pass is needed before concat_optional_stops.

Parameters:

• trips (DataFrame) – Combined Dask DataFrame of original and optional-stop trips, indexed by vehicle ID (output of get_optional_stop_trips).

• params (dict) –

  Pipeline parameters dict with keys:

  • columns (dict): sub-keys for dist_along_miles, speed_route, hours_orig, hours_route, start_time, routing_status, miles_route, miles_orig.

  • trip_id_cols (list[str]): columns that uniquely identify a trip (used for groupby and sort).

  • rename_cols_final (dict[str, str]): column renames applied at the end to restore original column names.

  • keep_cols_final (list[str]): columns to retain in the output.

Return type:

DataFrame

Returns:

A lazy dd.DataFrame of sub-trips with updated timestamps and distances, ready to be concatenated with the original trips.

laurel.pipelines.compute_routes.nodes.format_stop_locations(stops, params)

Convert truck-stop candidate records to a point GeoDataFrame for spatial joining.

Resets the index, applies column renames, creates point geometries from the H3 hexagon centroids, renames the geometry column, and assigns a contiguous integer stop ID.

Parameters:

• stops (DataFrame) – Raw truck-stop DataFrame (Jason’s Law or similar) indexed by hexagon ID.

• params (dict) –

  Pipeline parameters dict with keys:

  • columns (dict): sub-keys hex (hex-ID column), park_point (output geometry column name), park_id (output stop-ID column name).

  • col_renamer (dict[str, str]): mapping from raw to internal column names (inverted for renaming).

  • keep_cols (list[str]): columns to retain in the output.

Return type:

GeoDataFrame

Returns:

A gpd.GeoDataFrame with one row per truck-stop candidate and a point geometry column named params["columns"]["park_point"].

laurel.pipelines.compute_routes.nodes.get_optional_stop_trips(routes, parks, params)

Identify truck stops along each route and compute their distance from the trip origin.

Fuses, into a single per-partition task, everything that used to be a chain of separate Dask operations: dropping unused columns, falling back to pre-routing trip_miles/trip_hrs for trips with no route geometry, buffering and spatially joining truck stops onto the route LineStrings, projecting each matched stop onto its route to get its distance from the trip origin, dropping the now-unneeded route geometry, and excluding optional stops too close to either trip endpoint. Returns a lazy dd.DataFrame – nothing is computed here; the pipeline’s only materialization point is the final catalog write.

Parameters:

• routes (GeoDataFrame) – Routed trips Dask GeoDataFrame with route LineString geometry.

• parks (GeoDataFrame) – Truck-stop GeoDataFrame (output of format_stop_locations).

• params (dict) –

  Pipeline parameters dict with keys:

  • columns (dict): sub-keys for column names including route_geom, park_point, park_id, hex_end, hex_park, dist_along_miles, routing_status.

  • projected_crs (str | CRS): CRS used for buffering and distance projection.

  • park_buffer_miles (float): buffer radius around each truck stop (miles); also used to exclude stops adjacent to either trip endpoint.

  • drop_cols_initial (list[str]): columns to drop from the routes DataFrame before joining.

Return type:

DataFrame

Returns:

A lazy dd.DataFrame combining original trips (routing_status="original" or "unrouted") and optional truck-stop trips (routing_status="routed"), with a dist_along_miles column recording each record’s distance from the trip origin.

laurel.pipelines.compute_routes.nodes.get_routes_node(trips, server, params)

Compute shortest-path routes for all trips and convert units to miles and hours.

Calls get_routes on each Dask partition via map_partitions, forwarding the origin and destination geometry columns and GraphHopper client parameters. After routing, raw metric units are converted: distance from metres to miles, duration from seconds to hours, and speed is derived as miles per hour. The route LineString column becomes the active geometry.

Parameters:

• trips (GeoDataFrame) – Trips Dask GeoDataFrame with origin/destination point columns (output of partition_trips).

• server (GraphhopperContainerRouter) – Running GraphhopperContainerRouter context manager providing server.base_url.

• params (dict) –

  Pipeline parameters dict with keys:

  • input_cols (dict): sub-keys orig and dest naming the origin and destination geometry columns.

  • client (dict): max_concurrent_requests, batch_size, timeout_secs, verbose — forwarded to get_routes.

  • profile (str): GraphHopper vehicle profile.

  • output_trip_cols (dict): sub-keys dist, dur, speed naming the output columns.

Return type:

GeoDataFrame

Returns:

A Dask GeoDataFrame with route LineString geometry and added columns for route distance (miles), duration (hours), and speed (mph).

laurel.pipelines.compute_routes.nodes.get_trip_orig_dest_points(trips, params)

Convert origin and destination H3 hexagons to point geometries for routing.

Maps each H3 hex-ID column to a GeoSeries of centroid points using cells_to_points, then sets the active geometry to the output geometry column.

Parameters:

• trips (DataFrame) – Filtered trips Dask DataFrame with H3 hex-ID columns.

• params (dict) –

  Pipeline parameters dict with keys:

  • hex_geo_cols (dict[str, str]): mapping from output geometry column name to the source hex-ID column name (e.g., {"origin_geom": "origin_hex"}).

  • output_geom_col (str): name of the active geometry column to set on the output GeoDataFrame.

Return type:

GeoDataFrame

Returns:

A Dask GeoDataFrame with point geometry columns for origin and destination.

laurel.pipelines.compute_routes.nodes.import_graph(osm_path, server_params)

Import an OSM road-network file into a GraphHopper Docker container.

Starts a GraphhopperContainerRouter in import mode, which instructs the container to read the PBF/OSM file and build a routing graph on disk. This step only needs to be run once per road-network file; subsequent pipeline runs reuse the pre-built graph.

Parameters:

• osm_path (str) – Path to the OSM/PBF input file, supplied by the osm_north_america catalog entry. It arrives fully resolved, which matters because the container gets it verbatim and cannot expand environment variables such as $SCRATCH.

• server_params (dict) –

  GraphHopper server configuration dict with keys:

  • image (str): Docker image name/tag.

  • graph_dir (str): host path to the directory where the routing graph will be stored.

  • config_path (str): path to the GraphHopper config file.

  • resources (dict): sub-key import with mem_max_gb, mem_start_gb, and startup_delay_secs.

Return type:

None

laurel.pipelines.compute_routes.nodes.index_by_vehicle(trips, params)

Index a trips DataFrame by vehicle ID with known, monotonic divisions.

Used at two points in this pipeline where the caller needs a collection with real divisions on the vehicle-ID column so it can later be combined with another such collection via dd.concat(..., interleave_partitions=True) – a partition-wise merge rather than a full hash shuffle. An unsorted set_index does its own divisions-sampling pass and its own data-transfer pass over its input, so this is only called on inputs that are either cheap (no expensive ancestry to redo twice) or already checkpointed to disk (severing any expensive ancestry beforehand).

Parameters:

• trips (DataFrame) – Dask DataFrame to index.

• params (dict) –

  Pipeline parameters dict with keys:

  • id_col (str): vehicle-ID column to index by.

  • n_partitions (int): target output partition count.

Return type:

DataFrame

Returns:

trips indexed by id_col, with known divisions.

laurel.pipelines.compute_routes.nodes.partition_trips(trips, params)

Re-partition the trips GeoDataFrame before writing to disk.

Parameters:

• trips (GeoDataFrame) – Trips Dask GeoDataFrame with origin/destination geometries.

• params (dict) –

  Pipeline parameters dict with keys:

  • n_partitions (int): target number of Dask partitions.

Return type:

GeoDataFrame

Returns:

The GeoDataFrame repartitioned to params["n_partitions"] parts.

laurel.pipelines.compute_routes.nodes.select_trips_to_route(trips, params)

Split trips into those worth routing and those too short to bother.

Very short trips (below min_dist_miles) are excluded from routing because it would add noise without meaningfully changing the set of reachable truck stops – they can never gain an optional stop, so they are returned separately rather than dropped, ready to be reattached unchanged by concat_optional_stops further downstream. An optional debug subsample further reduces the to-route set for rapid iteration; it is not applied to the not-to-route set, since that set isn’t part of the expensive routing path debug_subsample exists to shrink.

trips is expected to already be indexed by vehicle ID (see index_by_vehicle) – both returned frames inherit that index and its divisions for free via .loc[], so nothing downstream needs to re-index the not-to-route side.

Parameters:

• trips (DataFrame) – Dask DataFrame of formatted trip records, indexed by vehicle ID.

• params (dict) –

  Pipeline parameters dict with keys:

  • drop_cols (list[str]): columns to remove before routing.

  • dist_col (str): trip-distance column name (miles).

  • min_dist_miles (float): minimum trip distance to retain.

  • debug_subsample (dict): active (bool) and frac (float) for fractional subsampling.

Return type:

tuple[DataFrame, DataFrame]

Returns:

A (to_route, not_to_route) tuple of Dask DataFrames.

laurel.pipelines.compute_routes.nodes.test_get_routes(route_params, server_params)

Send a health-check routing query to verify the GraphHopper server is operational.

Issues a single coast-to-coast route request (Vermont to California) and logs the returned distance. The pipeline should be halted if this node fails, as it indicates the routing server is unavailable.

Parameters:

• route_params (dict) –

  Route configuration dict with keys:

  • profile (str): GraphHopper vehicle profile (e.g., "car" or "truck").

• server_params (dict) – GraphHopper server configuration dict (same structure as import_graph); the resources.server sub-key is used.

Return type:

None

laurel.pipelines.compute_routes.pipeline module

Kedro pipeline definition for the compute_routes pipeline.

Wires the nodes from laurel.pipelines.compute_routes.nodes into a single Pipeline object. For full documentation of each node’s inputs, outputs, and algorithm, see laurel.pipelines.compute_routes.nodes.

Sub-pipelines / tags

• import — imports the OSM road network into the GraphHopper Docker container (run once; reuses the pre-built graph on subsequent runs).

• pre_routing — indexes trips by vehicle ID, splits off trips too short to route, converts H3 hex origins/destinations to point geometries, and re-partitions for checkpointing.

• routing — starts a Dask cluster and the GraphHopper server, computes shortest-path routes for all trips, then tears both down.

• insert_optional_stops — spatially joins truck-stop candidates onto routes, projects stop positions along each route, splits trips at stops, and concatenates the result with the never-routed trips. The vehicle-ID index (and its divisions) set in pre_routing survives unshuffled all the way through, recovered for free off the checkpoint dataset rather than re-established with a second index_by_vehicle pass.

To visualise the node graph interactively, run:

kedro viz run

then open http://localhost:4141 in a browser and select compute_routes from the pipeline dropdown.

laurel.pipelines.compute_routes.pipeline.create_pipeline(**kwargs)

Return type:

Pipeline

Module contents

This is a boilerplate pipeline ‘compute_routes’ generated using Kedro 0.19.3

laurel.pipelines.compute_routes.create_pipeline(**kwargs)

Return type:

Pipeline