nird package¶
NIRD - National Infrastructure Resilience Demonstrator
Submodules¶
nird.constants module¶
Constants used in the network flow model
nird.road module¶
nird.road_capacity module¶
nird.road_functions module¶
nird.road_osm module¶
Road Network Flow Model - Functions
- nird.road_osm.compute_edge_costs(path: List[int]) Tuple[float, float, float][source]¶
Calculate the total travel cost for the path
Parameters¶
- path: List
A list of edge indexes that define the path.
Returns¶
- od_voc: float
Vehicle operating costs of each trip.
- od_vot: float
Value of time of each trip.
- od_toll: float
Toll costs of each trip.
- nird.road_osm.cost_func(time: float, distance: float, voc: float, toll: float) Tuple[float, float, float][source]¶
Calculate the total travel cost.
Parameters¶
- time: float
Travel time: hour
- distance: float
Travel distance: mile
- voc:
Vehicle operating cost: £/km
Returns¶
- cost: float
The total travel costs: £
- c_time: float
The time-equivalent costs: £
- c_operate: float
The vehicle operating costs/fuel costs: £
- nird.road_osm.create_igraph_network(road_links: GeoDataFrame, road_nodes: GeoDataFrame) Tuple[Graph, Dict[str, float], Dict[str, float], Dict[str, float], Dict[str, float], DataFrame][source]¶
Create an undirected igraph network.
Parameters¶
road_links: gpd.GeoDataFrame road_nodes: gpd.GeoDataFrame initialSpeeds: dict
The free-flow speed of different types of road links.
Returns¶
- igraph.Graph
The road network.
- edge_cost_dict,
Total travel cost of each edge.
- edge_timecost_dict,
The time-equivalent cost of each edge.
- edge_operatecost_dict,
The fuel cost of each edge.
- nird.road_osm.create_urban_mask(etisplus_urban_roads: GeoDataFrame) GeoDataFrame[source]¶
To extract urban areas across Great Britain (GB) based on ETISPLUS datasets.
Parameters¶
- etisplus_urban_roads: gpd.GeoDataFrame
D6 ETISplus Roads (2010)
Returns¶
- urban_mask: gpd.GeoDataFrame
A binary file that spatially identify the urban areas across GB with values == 1.
- nird.road_osm.edge_init(road_links: GeoDataFrame, initial_capacity_dict: Dict[str, float], free_flow_speed_dict: Dict[str, float], urban_flow_speed_dict: Dict[str, float], min_flow_speed_dict: Dict[str, float], max_flow_speed_dict: Dict[str, float]) GeoDataFrame[source]¶
Network edges initialisation.
Parameters¶
road_links: gpd.GeoDataFrame initial_capacity_dict: Dict
The average daily capacity of road links (passengers per lane per day)
- free_flow_speed_dict: Dict
The free-flow edge speed of road links (mph).
- urban_flow_speed_dict: Dict
The maximum vehicle operating speed restriction in urban areas (mph).
- min_flow_speed_dict: Dict
The minimum vehicle operating speeds (mph).
- max_flow_speed_dict: Dict
The maximum flow speed of the flooded road links (mph).
Returns¶
- road_links: pd.DataFrame
with added attributes of initial edge flow, capacity, and speed.
- initial_speed_dict: Dict
initial vehicle operating speed of existing road links.
- nird.road_osm.edge_initial_speed_func(road_links: DataFrame, free_flow_speed_dict: Dict[str, float], urban_flow_speed_dict: Dict[str, float], min_flow_speed_dict: Dict[str, float], max_flow_speed_dict: Dict[str, float] = None) Tuple[DataFrame, Dict[str, float]][source]¶
Calculate the initial vehicle speed for network edges.
Parameters¶
- road_links: pd.DataFrame
network link features
- free_flow_speed_dict: Dict
free-flow vehicle operating speed on roads: M, A(single/dual), B
- urban_flow_speed_dict: Dict
set the maximum vehicle speed restrictions in urban areas
- min_flow_speed_dict: Dict
minimum vehicle operating speeds
- max_flow_speed_dict: Dict
maximum safe vehicel operatin speeds on flooded roads
Returns¶
- road_links: pd.DataFrame
with speed information
- initial_speed_dict: Dict
initial vehicle operating speed of existing road links
- nird.road_osm.extract_od_pairs(od: DataFrame) Tuple[List[str], Dict[str, List[str]], Dict[str, List[int]]][source]¶
Prepare the OD matrix.
Parameters¶
- od: pd.DataFrame
Table of origin-destination passenger flows.
Returns¶
- list_of_origin_nodes: list
A list of origin nodes.
- dict_of_destination_nodes: dict[str, list[str]]
A dictionary recording a list of destination nodes for each origin node.
- dict_of_origin_supplies: dict[str, list[int]]
A dictionary recording a list of flows for each origin-destination pair.
- nird.road_osm.find_least_cost_path(params: Tuple) Tuple[int, List[str], List[int], List[float]][source]¶
Find the least-cost path for each OD trip.
Parameters:¶
- params: Tuple
The first element: the origin node (index). The second element: a list of destination nodes (indexes). The third element: a list of outbound trips from origin to its connected destinations.
Returns:¶
- idx_of_origin_node: int
Same as input.
- list_of_idx_destination_nodes: list
Same as input.
- paths: list
The least-cost paths.
- flows: list
Same as input.
- nird.road_osm.find_nearest_node(zones: GeoDataFrame, road_nodes: GeoDataFrame) Dict[str, str][source]¶
Find the nearest road node for each admin unit.
Parameters¶
- zones: gpd.GeoDataFrame
Admin units.
- road nodes: gpd.GeoDataFrame
Nodes of the road network.
Returns¶
- nearest_node_dict: dict
A dictionary to convert from admin units to their attached road nodes.
- nird.road_osm.label_urban_roads(road_links: GeoDataFrame, urban_mask: GeoDataFrame) GeoDataFrame[source]¶
Classify road links into urban/suburban roads.
Parameters¶
- road_links: gpd.GeoDataFrame
Links of the road network.
- urban_mask: gpd.GeoDataFrame
A binary file that spatially identify the urban areas across GB.
Returns¶
- road_links: gpd.GeoDataFrame
Create a column “urban” to classify road links into urban/suburban links.
- nird.road_osm.network_flow_model(road_links: GeoDataFrame, road_nodes: GeoDataFrame, list_of_origins: List[str], supply_dict: Dict[str, List[float]], destination_dict: Dict[str, List[str]], free_flow_speed_dict: Dict[str, float], flow_breakpoint_dict: Dict[str, float], flow_capacity_dict: Dict[str, float], min_speed_cap: Dict[str, float], urban_speed_cap: Dict[str, float], max_flow_speed_dict=None) Tuple[DataFrame, DataFrame, Dict[Tuple[str, str], float]][source]¶
Model the passenger flows on the road network.
Parameters¶
road_links: gpd.GeoDataFrame road_nodes: gpd.GeoDataFrame list_of_origins: list
A list of unique origin nodes of OD flows.
- supply_dict: dict
The number of outbound trips of individual origins.
- destination_dict: dict
A list of destinations attached to individual origins.
- free_flow_speed_dict: dict
The free-flow speed of different types of road links.
- flow_breakpoint_dict: dict
The breakpoint flow of different types of road links, beyond which the flow speeds starts to decrease according to the remaining road capacities.
- flow_capacity_dict: dict
The capacity of different types of road links.
- max_flow_speed_dict: dict
The maximum vehicle speed of different flooded road links.
- min_speed_cap: dict
The restriction on the lowest flow rate.
- urban_speed_cap: dict
The restriction on the maximum flow rate in urban areas.
- max_flow_speed_dict: Dict
The maximum vehicle operating speeds.
Returns¶
- road_links: pd.DataFrame
The roak link features with simulating results of edge flows.
- isolated_flow_dict: dict
The isolated trips between each OD pair.
- odpfc: pd.DataFrame
The full od outputs: origins, destinations, paths, flows, costs.
- nird.road_osm.select_partial_roads_osm(road_links: GeoDataFrame, road_nodes: GeoDataFrame, col_name: str, list_of_values: List[str]) Tuple[GeoDataFrame, GeoDataFrame][source]¶
- nird.road_osm.speed_flow_func(asset_type: str, total_flow: float, initial_flow_speed: float, min_flow_speed: float, flow_breakpoint_dict: Dict[str, float]) float[source]¶
Create the speed-flow curves for different types of roads.
Parameters¶
- asset_type: str
The type or road links.
- total_flow: float
The number of vehicles on road links.
- initial_flow_speed: float
The initial traffic speeds of road links.
- min_flow_speed: float
The minimum traffic speeds of road links.
- flow_breakpoint_dict: dict
The breakpoint flows after which traffic flows starts to decrease.
Returns¶
- speed: float
The “real-time” traffic speeds on road links.
- nird.road_osm.update_network_structure(network: Graph, length_dict: Dict[str, float], speed_dict: Dict[str, float], toll_dict: Dict[str, float], temp_edge_flow: DataFrame) Tuple[Graph, Dict[str, float], Dict[str, float], Dict[str, float]][source]¶
Update the road network structure by removing the fully utilised edges and updating the weights of the remaining edges.
Parameters¶
- network: igraph.Graph
A road network.
- length_dict: dict
The length of road links.
- speed_dict:
The average flow speed of road links.
- temp_edge_flow: pd.DataFrame
A table that records edge flows.
Returns¶
- network: igraph.Graph
Th updated road network.
- edge_cost_dict: dict
The updated travel cost of edges.
- edge_timecost_dict: dict
The updated time-equivalent cost of edges.
- edge_operatecost_dict: dict
The updated vehicle operating cost of edges.
- nird.road_osm.update_od_matrix(temp_flow_matrix: DataFrame, supply_dict: Dict[str, List[float]], destination_dict: Dict[str, List[str]], isolated_flow_dict: Dict[Tuple[str, str], float]) Tuple[DataFrame, List[str], Dict[str, List[float]], Dict[str, List[str]]][source]¶
Update the OD matrix by removing unreachable desitinations from each origin; and origins with zero supplies.
Parameters¶
- temp_flow_matrix: pd.DataFrame
A temporary flow matrix: [origins, destinations, paths, flows]
- supply_dict: dict
The number of outbound trips from each origin.
- destination_dict: dict
A list of destinations attached to each origin.
- isolated_flow_dict: dict
The number of isolated trips between each OD pair.
Returns¶
temp_flow_matrix: pd.DataFrame new_list_of_origins: list new_supply_dict: dict new_destination: dict
- nird.road_osm.voc_func(speed: float) float[source]¶
Calculate the Vehicle Operating Cost (VOC).
Parameters¶
- speed: float
The average flow speed: mph
Returns¶
- float
The unit vehicle operating cost: £/km
- nird.road_osm.worker_init_edge(shared_network_pkl: bytes, shared_weight_pkl: bytes) None[source]¶
Worker initialisation in multiprocesses to create a shared network that could be used across different workers.
Parameters¶
- shared_network_pkl: bytes
The pickled file of the igraph network.
- shared_weight_pkl: bytes
The pickled flle of a dictionary of network edge weights.
Returns¶
None.
nird.road_recovery module¶
Road Network Flow Model - Functions
- nird.road_recovery.calculate_path(tfm_chunk: DataFrame) Tuple[DataFrame, DataFrame][source]¶
Calculate path attributes for a chunk of the flow matrix.
- nird.road_recovery.chunk_df(df, n_chunks: int)[source]¶
Split dataframe into n roughly equal chunks.
- nird.road_recovery.cost_func(time: float, distance: float, voc_per_km: float, toll: float) Tuple[float, float, float][source]¶
Calculate the total travel cost.
Parameters¶
- time: float
Travel time: hour
- distance: float
Travel distance: mile
- voc:
Vehicle operating cost: £/km
Returns¶
- cost: float
The total travel costs: £
- c_time: float
The time-equivalent costs: £
- c_operate: float
The vehicle operating costs/fuel costs: £
- nird.road_recovery.create_igraph_network(road_links: GeoDataFrame) Graph[source]¶
Create an undirected igraph network.
- nird.road_recovery.create_urban_mask(etisplus_urban_roads: GeoDataFrame) GeoDataFrame[source]¶
To extract urban areas across Great Britain (GB) based on ETISPLUS datasets.
Parameters¶
- etisplus_urban_roads: gpd.GeoDataFrame
D6 ETISplus Roads (2010)
Returns¶
- urban_mask: gpd.GeoDataFrame
A binary file that spatially identify the urban areas across GB with values == 1.
- nird.road_recovery.edge_init(road_links: GeoDataFrame, flow_breakpoint_dict: Dict[str, float], capacity_plph_dict: Dict[str, float], free_flow_speed_dict: Dict[str, float], urban_flow_speed_dict: Dict[str, float], min_flow_speed_dict: Dict[str, float], max_flow_speed_dict: Dict[str, float]) GeoDataFrame[source]¶
Network edges initialisation.
Parameters¶
road_links: gpd.GeoDataFrame capacity_plph_dict: Dict
The designed edge capacity (per lane per hour).
- free_flow_speed_dict: Dict
The free-flow edge speed of different types of road links.
- urban_flow_speed_dict: Dict
The maximum vehicle operating speed restriction in urban areas.
- min_flow_speed_dict: Dict
The minimum vehicle operating speeds.
- max_flow_speed_dict: Dict
The maximum flow speed of the flooded road links.
Returns¶
- road_links: gpd.GeoDataFrame
Road links with added attributes.
- nird.road_recovery.edge_initial_speed_func(road_links: DataFrame, flow_breakpoint_dict: Dict[str, float], free_flow_speed_dict: Dict[str, float], urban_flow_speed_dict: Dict[str, float], min_flow_speed_dict: Dict[str, float], max_flow_speed_dict: Dict[str, float] = None) Tuple[DataFrame, Dict[str, float]][source]¶
Calculate the initial vehicle speed for network edges.
Parameters¶
- road_links: pd.DataFrame
network link features
- flow_breakpoint_dict: Dict
the breakpoint flow beyond which vehicle speed start to decrease
- free_flow_speed_dict: Dict
free-flow vehicle operating speed on roads: M, A(single/dual), B
- urban_flow_speed_dict: Dict
set the maximum vehicle speed restrictions in urban areas
- min_flow_speed_dict: Dict
minimum vehicle operating speeds
- max_flow_speed_dict: Dict
maximum safe vehicel operatin speeds on flooded roads
Returns¶
- road_links: pd.DataFrame
with speed information
- initial_speed_dict: Dict
initial vehicle operating speed of existing road links
- nird.road_recovery.edge_reclassification_func(road_links: DataFrame) DataFrame[source]¶
Reclassify network edges to “M, A_dual, A_single, B”.
- nird.road_recovery.extract_od_pairs(od: DataFrame) Tuple[List[str], Dict[str, List[str]], Dict[str, List[int]]][source]¶
Prepare the OD matrix.
Parameters¶
- od: pd.DataFrame
Table of origin-destination passenger flows.
Returns¶
- list_of_origin_nodes: list
A list of origin nodes.
- dict_of_destination_nodes: dict[str, list[str]]
A dictionary recording a list of destination nodes for each origin node.
- dict_of_origin_supplies: dict[str, list[int]]
A dictionary recording a list of flows for each origin-destination pair.
- nird.road_recovery.find_least_cost_path(params: Tuple) Tuple[int, List[str], List[int], List[float]][source]¶
Find the least-cost path for each OD trip.
Parameters:¶
- params: Tuple
The first element: the origin node (index). The second element: a list of destination nodes (indexes). The third element: a list of outbound trips from origin to its connected
destinations.
Returns:¶
- idx_of_origin_node: int
Same as input.
- list_of_idx_destination_nodes: list
Same as input.
- paths: list
The least-cost paths.
- flows: list
Same as input.
- nird.road_recovery.find_nearest_node(zones: GeoDataFrame, road_nodes: GeoDataFrame, zone_id_column: str, node_id_column: str) Dict[str, str][source]¶
Find the nearest road node for each admin unit.
Parameters¶
- zones: gpd.GeoDataFrame
Admin units.
- road nodes: gpd.GeoDataFrame
Nodes of the road network.
Returns¶
- nearest_node_dict: dict
A dictionary to convert from admin units to their attached road nodes.
- nird.road_recovery.itter_path(network, road_links, temp_flow_matrix: DataFrame, num_of_chunk: int = None) Tuple[DataFrame, DataFrame][source]¶
Iterate through all the paths to calculate edge flows and travel costs.
- nird.road_recovery.label_urban_roads(road_links: GeoDataFrame, urban_mask: GeoDataFrame) GeoDataFrame[source]¶
Classify road links into urban/suburban roads.
Parameters¶
- road_links: gpd.GeoDataFrame
Links of the road network.
- urban_mask: gpd.GeoDataFrame
A binary file that spatially identify the urban areas across GB.
Returns¶
- road_links: gpd.GeoDataFrame
Create a column “urban” to classify road links into urban/suburban links.
- nird.road_recovery.network_flow_model(road_links: GeoDataFrame, network: Graph, remain_od: DataFrame, flow_breakpoint_dict: Dict[str, float], num_of_chunk: int, num_of_cpu: int) Tuple[GeoDataFrame, List, List][source]¶
Process-based Network Flow Simulation.
Parameters¶
road_links: road links network: igraph network remain_od: od matrix flow_breakpoint_dict: flow breakpoints of different types of road links
Returns¶
road_links: with added attributes (acc_flow, acc_capacity, acc_speed) isolation: non-allocated od matrix odpfc: allocated od matrix
- nird.road_recovery.retrieve_path_attributes(path: List[int]) Dict[source]¶
Retrieve attributes for a given path.
- nird.road_recovery.select_partial_roads(road_links: GeoDataFrame, road_nodes: GeoDataFrame, col_name: str, list_of_values: List[str]) Tuple[GeoDataFrame, GeoDataFrame][source]¶
Extract partial road network based on road types.
Parameters¶
- road_links: gpd.GeoDataFrame
Links of the road network.
- road_nodes: gpd.GeoDataFrame
Nodes of the road network.
- col_name: str
The road type column.
- list_of_values:
The road types to be extracted from the network.
Returns¶
- selected_links: gpd.GeoDataFrame
Partial road links.
- selected_nodes: gpd.GeoDataFrame
Partial road nodes.
- nird.road_recovery.update_edge_speed(combined_label: str, total_flow: float, initial_flow_speed: float, min_flow_speed: float, breakpoint_flow: float) float[source]¶
Create the speed-flow curves for different types of roads.
Parameters¶
- combined_label: str
The type or road links.
- total_flow: float
The number of vehicles on road links (passneger-cars).
- initial_flow_speed: float
The initial traffic speeds of road links (mph).
- min_flow_speed: float
The minimum traffic speeds of road links (mph).
- breakpoint_flow: float
The breakpoint flow after which speed start to decrease. (passenger-cars/hour/lane)
Returns¶
- speed: float
The “real-time” traffic speeds on road links. (mph)
- nird.road_recovery.update_network_structure(num_of_edges: int, network: Graph, temp_edge_flow: DataFrame, road_links: GeoDataFrame) Graph[source]¶
Drop fully utilised edges and Update edge weights.
Parameters¶
network: igraph network temp_edge_flow: the remaining edge capacity at the current iteration road_links: road links
Returns¶
The updated igraph network
- nird.road_recovery.update_od_matrix(temp_flow_matrix: DataFrame) Tuple[DataFrame, DataFrame, DataFrame][source]¶
Divide the OD matrix into allocated and unallocated sections.
Parameters¶
- temp_flow_matrix: origin, destination, path, flow, operating_cost_per_flow,
time_cost_per_flow, toll_cost_per_flow
remain_od: origin, destination, flow
Returns¶
temp_flow_matrix: flow matrix with valid path isolated_flow_matrix: flow matrix with no path remain_od: the remaining od matrix
nird.road_revised module¶
nird.road_revised_recovery module¶
Road Network Flow Model - Functions
- nird.road_revised_recovery.cost_func(time: float, distance: float, voc_per_km: float, toll: float) Tuple[float, float, float][source]¶
Calculate the total travel cost.
Parameters¶
- time: float
Travel time: hour
- distance: float
Travel distance: mile
- voc:
Vehicle operating cost: £/km
Returns¶
- cost: float
The total travel costs: £
- c_time: float
The time-equivalent costs: £
- c_operate: float
The vehicle operating costs/fuel costs: £
- nird.road_revised_recovery.create_igraph_network(road_links: GeoDataFrame) Graph[source]¶
Create an undirected igraph network.
- nird.road_revised_recovery.create_urban_mask(etisplus_urban_roads: GeoDataFrame) GeoDataFrame[source]¶
To extract urban areas across Great Britain (GB) based on ETISPLUS datasets.
Parameters¶
- etisplus_urban_roads: gpd.GeoDataFrame
D6 ETISplus Roads (2010)
Returns¶
- urban_mask: gpd.GeoDataFrame
A binary file that spatially identify the urban areas across GB with values == 1.
- nird.road_revised_recovery.edge_init(road_links: GeoDataFrame, flow_breakpoint_dict: Dict[str, float], capacity_plph_dict: Dict[str, float], free_flow_speed_dict: Dict[str, float], urban_flow_speed_dict: Dict[str, float], min_flow_speed_dict: Dict[str, float], max_flow_speed_dict: Dict[str, float]) GeoDataFrame[source]¶
Network edges initialisation.
Parameters¶
road_links: gpd.GeoDataFrame capacity_plph_dict: Dict
The designed edge capacity (per lane per hour).
- free_flow_speed_dict: Dict
The free-flow edge speed of different types of road links.
- urban_flow_speed_dict: Dict
The maximum vehicle operating speed restriction in urban areas.
- min_flow_speed_dict: Dict
The minimum vehicle operating speeds.
- max_flow_speed_dict: Dict
The maximum flow speed of the flooded road links.
Returns¶
- road_links: gpd.GeoDataFrame
Road links with added attributes.
- nird.road_revised_recovery.edge_initial_speed_func(road_links: DataFrame, flow_breakpoint_dict: Dict[str, float], free_flow_speed_dict: Dict[str, float], urban_flow_speed_dict: Dict[str, float], min_flow_speed_dict: Dict[str, float], max_flow_speed_dict: Dict[str, float] = None) Tuple[DataFrame, Dict[str, float]][source]¶
Calculate the initial vehicle speed for network edges.
Parameters¶
- road_links: pd.DataFrame
network link features
- flow_breakpoint_dict: Dict
the breakpoint flow beyond which vehicle speed start to decrease
- free_flow_speed_dict: Dict
free-flow vehicle operating speed on roads: M, A(single/dual), B
- urban_flow_speed_dict: Dict
set the maximum vehicle speed restrictions in urban areas
- min_flow_speed_dict: Dict
minimum vehicle operating speeds
- max_flow_speed_dict: Dict
maximum safe vehicel operatin speeds on flooded roads
Returns¶
- road_links: pd.DataFrame
with speed information
- initial_speed_dict: Dict
initial vehicle operating speed of existing road links
- nird.road_revised_recovery.edge_reclassification_func(road_links: DataFrame) DataFrame[source]¶
Reclassify network edges to “M, A_dual, A_single, B”.
- nird.road_revised_recovery.extract_od_pairs(od: DataFrame) Tuple[List[str], Dict[str, List[str]], Dict[str, List[int]]][source]¶
Prepare the OD matrix.
Parameters¶
- od: pd.DataFrame
Table of origin-destination passenger flows.
Returns¶
- list_of_origin_nodes: list
A list of origin nodes.
- dict_of_destination_nodes: dict[str, list[str]]
A dictionary recording a list of destination nodes for each origin node.
- dict_of_origin_supplies: dict[str, list[int]]
A dictionary recording a list of flows for each origin-destination pair.
- nird.road_revised_recovery.find_least_cost_path(params: Tuple) Tuple[int, List[str], List[int], List[float]][source]¶
Find the least-cost path for each OD trip.
Parameters:¶
- params: Tuple
The first element: the origin node (index). The second element: a list of destination nodes (indexes). The third element: a list of outbound trips from origin to its connected
destinations.
Returns:¶
- idx_of_origin_node: int
Same as input.
- list_of_idx_destination_nodes: list
Same as input.
- paths: list
The least-cost paths.
- flows: list
Same as input.
- nird.road_revised_recovery.find_nearest_node(zones: GeoDataFrame, road_nodes: GeoDataFrame, zone_id_column: str, node_id_column: str) Dict[str, str][source]¶
Find the nearest road node for each admin unit.
Parameters¶
- zones: gpd.GeoDataFrame
Admin units.
- road nodes: gpd.GeoDataFrame
Nodes of the road network.
Returns¶
- nearest_node_dict: dict
A dictionary to convert from admin units to their attached road nodes.
- nird.road_revised_recovery.itter_path(network, temp_flow_matrix: DataFrame) Tuple[DataFrame, DataFrame][source]¶
Iterate through all the paths to calculate edge flows and travel costs.
- nird.road_revised_recovery.label_urban_roads(road_links: GeoDataFrame, urban_mask: GeoDataFrame) GeoDataFrame[source]¶
Classify road links into urban/suburban roads.
Parameters¶
- road_links: gpd.GeoDataFrame
Links of the road network.
- urban_mask: gpd.GeoDataFrame
A binary file that spatially identify the urban areas across GB.
Returns¶
- road_links: gpd.GeoDataFrame
Create a column “urban” to classify road links into urban/suburban links.
- nird.road_revised_recovery.network_flow_model(road_links: GeoDataFrame, network: Graph, remain_od: DataFrame, flow_breakpoint_dict: Dict[str, float], num_of_cpu) Tuple[GeoDataFrame, List, List][source]¶
Process-based Network Flow Simulation.
Parameters¶
road_links: road links network: igraph network remain_od: od matrix flow_breakpoint_dict: flow breakpoints of different types of road links
Returns¶
road_links: with added attributes (acc_flow, acc_capacity, acc_speed) isolation: non-allocated od matrix odpfc: allocated od matrix
- nird.road_revised_recovery.select_partial_roads(road_links: GeoDataFrame, road_nodes: GeoDataFrame, col_name: str, list_of_values: List[str]) Tuple[GeoDataFrame, GeoDataFrame][source]¶
Extract partial road network based on road types.
Parameters¶
- road_links: gpd.GeoDataFrame
Links of the road network.
- road_nodes: gpd.GeoDataFrame
Nodes of the road network.
- col_name: str
The road type column.
- list_of_values:
The road types to be extracted from the network.
Returns¶
- selected_links: gpd.GeoDataFrame
Partial road links.
- selected_nodes: gpd.GeoDataFrame
Partial road nodes.
- nird.road_revised_recovery.update_edge_speed(combined_label: str, total_flow: float, initial_flow_speed: float, min_flow_speed: float, breakpoint_flow: float) float[source]¶
Create the speed-flow curves for different types of roads.
Parameters¶
- combined_label: str
The type or road links.
- total_flow: float
The number of vehicles on road links.
- initial_flow_speed: float
The initial traffic speeds of road links.
- min_flow_speed: float
The minimum traffic speeds of road links.
- breakpoint_flow: float
The breakpoint flow after which speed start to decrease.
Returns¶
- speed: float
The “real-time” traffic speeds on road links.
- nird.road_revised_recovery.update_network_structure(num_of_edges: int, network: Graph, temp_edge_flow: DataFrame, road_links: GeoDataFrame) Graph[source]¶
Drop fully utilised edges and Update edge weights.
Parameters¶
network: igraph network temp_edge_flow: the remaining edge capacity at the current iteration road_links: road links
Returns¶
The updated igraph network
- nird.road_revised_recovery.update_od_matrix(temp_flow_matrix: DataFrame, remain_od: DataFrame) Tuple[DataFrame, DataFrame, DataFrame][source]¶
Divide the OD matrix into allocated and unallocated sections.
Parameters¶
- temp_flow_matrix: origin, destination, path, flow, operating_cost_per_flow,
time_cost_per_flow, toll_cost_per_flow
remain_od: origin, destination, flow
Returns¶
temp_flow_matrix: flow matrix with valid path isolated_flow_matrix: flow matrix with no path remain_od: the remaining od matrix
nird.road_validation module¶
Road Network Flow Model - Functions
- nird.road_validation.cost_func(time: float, distance: float, voc_per_km: float, toll: float) Tuple[float, float, float][source]¶
Calculate the total travel cost.
Parameters¶
- time: float
Travel time: hour
- distance: float
Travel distance: mile
- voc:
Vehicle operating cost: £/km
Returns¶
- cost: float
The total travel costs: £
- c_time: float
The time-equivalent costs: £
- c_operate: float
The vehicle operating costs/fuel costs: £
- nird.road_validation.create_igraph_network(road_links: GeoDataFrame) Graph[source]¶
Create an undirected igraph network.
- nird.road_validation.create_urban_mask(etisplus_urban_roads: GeoDataFrame) GeoDataFrame[source]¶
To extract urban areas across Great Britain (GB) based on ETISPLUS datasets.
Parameters¶
- etisplus_urban_roads: gpd.GeoDataFrame
D6 ETISplus Roads (2010)
Returns¶
- urban_mask: gpd.GeoDataFrame
A binary file that spatially identify the urban areas across GB with values == 1.
- nird.road_validation.edge_init(road_links: GeoDataFrame, capacity_plph_dict: Dict[str, float], free_flow_speed_dict: Dict[str, float], urban_flow_speed_dict: Dict[str, float], min_flow_speed_dict: Dict[str, float], max_flow_speed_dict: Dict[str, float]) GeoDataFrame[source]¶
Network edges initialisation.
Parameters¶
road_links: gpd.GeoDataFrame capacity_plph_dict: Dict
The designed edge capacity (per lane per hour).
- free_flow_speed_dict: Dict
The free-flow edge speed of different types of road links.
- urban_flow_speed_dict: Dict
The maximum vehicle operating speed restriction in urban areas.
- min_flow_speed_dict: Dict
The minimum vehicle operating speeds.
- max_flow_speed_dict: Dict
The maximum flow speed of the flooded road links.
Returns¶
- road_links: gpd.GeoDataFrame
Road links with added attributes.
- nird.road_validation.edge_initial_speed_func(road_links: DataFrame, free_flow_speed_dict: Dict[str, float], urban_flow_speed_dict: Dict[str, float], min_flow_speed_dict: Dict[str, float], max_flow_speed_dict: Dict[str, float] = None) Tuple[DataFrame, Dict[str, float]][source]¶
Calculate the initial vehicle speed for network edges.
Parameters¶
- road_links: pd.DataFrame
network link features
- free_flow_speed_dict: Dict
free-flow vehicle operating speed on roads: M, A(single/dual), B
- urban_flow_speed_dict: Dict
set the maximum vehicle speed restrictions in urban areas
- min_flow_speed_dict: Dict
minimum vehicle operating speeds
- max_flow_speed_dict: Dict
maximum safe vehicel operatin speeds on flooded roads
Returns¶
- road_links: pd.DataFrame
with speed information
- initial_speed_dict: Dict
initial vehicle operating speed of existing road links
- nird.road_validation.edge_reclassification_func(road_links: DataFrame) DataFrame[source]¶
Reclassify network edges to “M, A_dual, A_single, B”.
- nird.road_validation.extract_od_pairs(od: DataFrame) Tuple[List[str], Dict[str, List[str]], Dict[str, List[int]]][source]¶
Prepare the OD matrix.
Parameters¶
- od: pd.DataFrame
Table of origin-destination passenger flows.
Returns¶
- list_of_origin_nodes: list
A list of origin nodes.
- dict_of_destination_nodes: dict[str, list[str]]
A dictionary recording a list of destination nodes for each origin node.
- dict_of_origin_supplies: dict[str, list[int]]
A dictionary recording a list of flows for each origin-destination pair.
- nird.road_validation.find_least_cost_path(params: Tuple) Tuple[int, List[str], List[int], List[float]][source]¶
Find the least-cost path for each OD trip.
Parameters:¶
- params: Tuple
The first element: the origin node (index). The second element: a list of destination nodes (indexes). The third element: a list of outbound trips from origin to its connected destinations.
Returns:¶
- idx_of_origin_node: int
Same as input.
- list_of_idx_destination_nodes: list
Same as input.
- paths: list
The least-cost paths.
- flows: list
Same as input.
- nird.road_validation.find_nearest_node(zones: GeoDataFrame, road_nodes: GeoDataFrame, zone_id_column: str, node_id_column: str) Dict[str, str][source]¶
Find the nearest road node for each admin unit.
Parameters¶
- zones: gpd.GeoDataFrame
Admin units.
- road nodes: gpd.GeoDataFrame
Nodes of the road network.
Returns¶
- nearest_node_dict: dict
A dictionary to convert from admin units to their attached road nodes.
- nird.road_validation.label_urban_roads(road_links: GeoDataFrame, urban_mask: GeoDataFrame) GeoDataFrame[source]¶
Classify road links into urban/suburban roads.
Parameters¶
- road_links: gpd.GeoDataFrame
Links of the road network.
- urban_mask: gpd.GeoDataFrame
A binary file that spatially identify the urban areas across GB.
Returns¶
- road_links: gpd.GeoDataFrame
Create a column “urban” to classify road links into urban/suburban links.
- nird.road_validation.network_flow_model(road_links: GeoDataFrame, network: Graph, remain_od: DataFrame, flow_breakpoint_dict: Dict[str, float], num_of_cpu: int) Tuple[GeoDataFrame, List, List][source]¶
Process-based Network Flow Simulation.
Parameters¶
road_links: road links network: igraph network remain_od: od matrix flow_breakpoint_dict: flow breakpoints of different types of road links
Returns¶
road_links: with added attributes (acc_flow, acc_capacity, acc_speed) isolation: non-allocated od matrix odpfc: allocated od matrix
- nird.road_validation.select_partial_roads(road_links: GeoDataFrame, road_nodes: GeoDataFrame, col_name: str, list_of_values: List[str]) Tuple[GeoDataFrame, GeoDataFrame][source]¶
Extract partial road network based on road types.
Parameters¶
- road_links: gpd.GeoDataFrame
Links of the road network.
- road_nodes: gpd.GeoDataFrame
Nodes of the road network.
- col_name: str
The road type column.
- list_of_values:
The road types to be extracted from the network.
Returns¶
- selected_links: gpd.GeoDataFrame
Partial road links.
- selected_nodes: gpd.GeoDataFrame
Partial road nodes.
- nird.road_validation.speed_flow_func(combined_label: str, total_flow: float, initial_flow_speed: float, min_flow_speed: float, flow_breakpoint_dict: Dict[str, float]) float[source]¶
Create the speed-flow curves for different types of roads.
Parameters¶
- combined_label: str
The type or road links.
- total_flow: float
The number of vehicles on road links.
- initial_flow_speed: float
The initial traffic speeds of road links.
- min_flow_speed: float
The minimum traffic speeds of road links.
- flow_breakpoint_dict: dict
The breakpoint flows after which traffic flows starts to decrease.
Returns¶
- speed: float
The “real-time” traffic speeds on road links.
- nird.road_validation.update_network_structure(network: Graph, temp_edge_flow: DataFrame, road_links: GeoDataFrame) Graph[source]¶
Drop fully utilised edges and update edge weights.
Parameters¶
network: igraph network temp_edge_flow: the remaining edge capacity at the current iteration road_links: road links
Returns¶
The updated igraph network
- nird.road_validation.update_od_matrix(temp_flow_matrix: DataFrame, remain_od: DataFrame) Tuple[DataFrame, DataFrame, DataFrame][source]¶
Divide the OD matrix into allocated and unallocated sections.
Parameters¶
- temp_flow_matrix: origin, destination, path, flow, operating_cost_per_flow,
time_cost_per_flow, toll_cost_per_flow
remain_od: origin, destination, flow
Returns¶
temp_flow_matrix: flow matrix with valid path isolated_flow_matrix: flow matrix with no path remain_od: the remaining od matrix
nird.utils module¶
Utility functions
- nird.utils.add_lines(x: DataFrame, from_nodes_df: DataFrame, to_nodes_df: DataFrame, from_nodes_id: str, to_nodes_id: str) LineString[source]¶
- nird.utils.components(edges: DataFrame, nodes: DataFrame, node_id_col: str) Tuple[DataFrame, DataFrame][source]¶
- nird.utils.create_network_from_nodes_and_edges(nodes: GeoDataFrame | None, edges: GeoDataFrame, node_edge_prefix: str, snap_distance: float | None = None, by: List[str] | None = None) Network[source]¶
- nird.utils.extract_gdf_values_containing_nodes(x: GeoSeries, input_gdf: GeoDataFrame, column_name: str) GeoSeries[source]¶
- nird.utils.gdf_geom_clip(gdf_in: GeoDataFrame, clip_geom: Polygon) GeoDataFrame[source]¶
Filter a dataframe to contain only features within a clipping geometry
Parameters¶
- gdf_in
geopandas dataframe to be clipped in
- province_geom
shapely geometry of province for what we do the calculation
Returns¶
filtered dataframe
- nird.utils.getDistance(loc1: Tuple[float | int, float | int], loc2: Tuple[float | int, float | int]) float[source]¶
Haversine formula - give coordinates as (lat_decimal,lon_decimal) tuples
- nird.utils.get_flow_on_edges(save_paths_df: DataFrame, edge_id_column: str, edge_path_column: str, flow_column: str) DataFrame[source]¶
Get flows from paths onto edges Parameters ——— save_paths_df
Pandas DataFrame of OD flow paths and their flow values
- edge_id_column
String name of ID column of edge dataset
- edge_path_column
String name of column which contains edge paths
- flow_column
String name of column which contains flow values
Result¶
DataFrame with edge_ids and their total flows
- nird.utils.get_nearest_values(x: GeoSeries, input_gdf: GeoDataFrame, column_name: str) GeoDataFrame[source]¶
- nird.utils.load_config(config_path: str | None = None) Dict[str, Dict[str, str]][source]¶
Read config.json
- nird.utils.network_od_path_estimations(graph: Graph, source: str | float | int, target: str | float | int, cost_criteria: str) Tuple[List[List[str | float | int]], List[float]][source]¶
Estimate the paths, distances, times, and costs for given OD pair
Parameters¶
- graph
igraph network structure
- source
name of Origin node ID
- target
name of Destination node ID
- cost_criteria
name of generalised cost criteria to be used: min_gcost or max_gcost
Returns¶
- edge_path_list
nested lists of edge ID’s in routes
- path_gcost_list
estimated generalised costs of routes