"""Kedro pipeline definition for the ``electrify_trips`` pipeline.
Wires the nodes from :mod:`laurel.pipelines.electrify_trips.nodes` into a single ``Pipeline`` object.
For full documentation of each node's inputs, outputs, and algorithm,
see :mod:`laurel.pipelines.electrify_trips.nodes`.
Sub-pipelines / tags
--------------------
- **choose_charging** — the single ``charge_pipe`` that
runs the full electrification simulation for one State of the World:
filters vehicles, assigns battery ranges, calculates energy requirements,
marks critical days, and runs the utility-maximisation charging-choice
algorithm. Outputs are written to a scenario-keyed partition in
``data/07_model_output/``.
To visualise the node graph interactively, run::
kedro viz run
then open http://localhost:4141 in a browser and select ``electrify_trips``
from the pipeline dropdown.
"""
from kedro.pipeline import Node, Pipeline
from laurel.models.dwell_sets import load_dwell_set, save_dwell_set
from laurel.utils.data import categorize_columns, filter_by_vals_in_cols
from laurel.utils.params import set_entity_params
from .nodes import (
assign_modes,
calc_dwell_durations,
calc_energy_use,
calc_vehicle_ranges,
drop_cols_to_pandas,
filter_dwells,
filter_dwells_post,
filter_vehicles,
mark_critical_days,
mark_shift_powers,
merge_dwellset_node,
prepare_modes,
simulate_charging_choice,
)
[docs]
def create_pipeline(**kwargs) -> Pipeline:
charge_pipe = Pipeline(
[
Node(
func=filter_by_vals_in_cols,
inputs=["vehicles_labelled", "params:filter_vehicles"],
outputs="vehicles_filtered",
name="filter_by_vals_in_cols_vehs",
),
Node(
func=drop_cols_to_pandas,
inputs=["vehicles_filtered", "params:drop_veh_geom_cols"],
outputs="vehicles_filtered_pd",
name="drop_veh_geom_cols",
),
Node(
func=set_entity_params,
inputs=["vehicles_filtered_pd", "params:vehicles"],
outputs="vehicles_with_params",
name="set_entity_params_vehicles",
),
Node(
func=load_dwell_set,
inputs=["dwells_with_locations_dask", "params:load_dwell_set"],
outputs="dwell_obj",
name="load_dwell_set_electrify_trips",
),
Node(
func=filter_vehicles,
inputs=[
"dwell_obj",
"vehicles_with_params",
"params:filter_vehicles_partitions",
],
outputs="dwell_obj_filtered_vehs",
name="filter_vehicles",
),
Node(
func=categorize_columns,
inputs="dwell_obj_filtered_vehs",
outputs="dwell_obj_filtered_vehs_categorized",
name="categorize_columns_electrify_trips",
),
Node(
func=calc_vehicle_ranges,
inputs=[
"vehicles_with_params",
"dwell_obj", # Using original dwell_obj to avoid double-computing
"params:calc_vehicle_ranges",
],
outputs="vehicles_with_params_scenario",
name="calc_vehicle_ranges",
),
Node(
func=merge_dwellset_node,
inputs=[
"dwell_obj_filtered_vehs_categorized",
"vehicles_with_params_scenario",
"params:merge_vehicle_params",
],
outputs="dwell_obj_w_veh_params",
name="merge_dwellset_node_veh_params",
),
Node(
func=calc_dwell_durations,
inputs=[
"dwell_obj_w_veh_params",
"params:calc_dwell_durations",
],
outputs="dwell_obj_w_dur",
name="calc_dwell_durations",
tags="frame-spatiotemporal",
),
Node(
func=prepare_modes,
inputs="params:charging_modes",
outputs="charging_modes",
name="prepare_modes",
),
Node(
func=assign_modes,
inputs=["dwell_obj_w_dur", "charging_modes", "params:assign_modes"],
outputs="dwell_obj_w_avail_modes",
name="assign_modes",
),
Node(
func=calc_energy_use,
inputs=[
"dwell_obj_w_avail_modes",
"params:calc_energy_use",
],
outputs="dwell_obj_w_energy",
name="calc_energy_use",
tags="frame-energy",
),
Node(
func=mark_critical_days,
inputs=[
"dwell_obj_w_energy",
"params:mark_critical_days",
],
outputs="dwell_obj_crit_days",
name="mark_critical_days",
tags="frame-spatiotemporal",
),
Node(
func=filter_dwells,
inputs=["dwell_obj_crit_days", "params:filter_dwells"],
outputs="dwell_obj_crit_dwells",
name="filter_dwells",
tags="frame-spatiotemporal",
),
Node(
func=mark_shift_powers,
inputs=["dwell_obj_crit_dwells", "params:mark_shift_powers"],
outputs="dwell_obj_shift_powers",
name="mark_shift_powers",
tags="frame-charging_choice",
),
Node(
func=simulate_charging_choice,
inputs=[
"dwell_obj_shift_powers",
"vehicles_with_params_scenario",
"charging_modes",
"params:simulate_charging_choice",
],
outputs="dwell_obj_w_charging",
name="simulate_charging_choice",
tags="frame-charging_choice",
),
Node(
func=filter_dwells_post,
inputs=["dwell_obj_w_charging", "params:filter_dwells_post"],
outputs="dwell_obj_no_unused_optionals",
name="filter_dwells_post",
tags="frame-charging_choice",
),
Node(
func=merge_dwellset_node,
inputs=[
"dwell_obj_no_unused_optionals",
"charging_modes",
"params:merge_chosen_mode",
],
outputs="dwell_obj_w_modes",
name="merge_dwellset_node_chosen_mode",
),
Node(
func=save_dwell_set,
inputs="dwell_obj_w_modes",
outputs="dwells_with_charging_scenario_dask",
name="save_dwell_set",
),
],
tags=["choose_charging"],
)
return charge_pipe