nird package¶
NIRD - National Infrastructure Resilience Demonstrator
Submodules¶
nird.constants module¶
Constants used in the network flow model
nird.road 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
This module provides a set of functions to model and analyze road network flows. It includes utilities for road network initialization, traffic flow simulation, cost calculations, and network updates. The functions are designed to handle geospatial data and integrate with igraph for network analysis.
Key Features: - Road network extraction and classification. - Traffic speed and flow modeling. - Cost estimation for vehicle operations and travel time. - Network flow simulation with iterative updates. - Integration with multiprocessing for performance optimization.
- nird.road_recovery.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_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]¶
Generate a spatial mask for urban areas in Great Britain.
Parameters¶
- etisplus_urban_roads: gpd.GeoDataFrame
Geospatial data of ETISPLUS urban roads (2010 dataset).
Returns¶
- urban_mask: gpd.GeoDataFrame
A geospatial mask identifying urban areas with a binary value (1 for urban).
- nird.road_recovery.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_recovery.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_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.label_urban_roads(road_links: GeoDataFrame, urban_mask: GeoDataFrame) GeoDataFrame[source]¶
Label road links as urban or suburban based on spatial intersection.
Parameters¶
- road_links: gpd.GeoDataFrame
Geospatial data representing the links of the road network.
- urban_mask: gpd.GeoDataFrame
Geospatial mask identifying urban areas.
Returns¶
- road_links: gpd.GeoDataFrame
Updated road links with a new column “urban” (1 for urban, 0 for suburban).
- nird.road_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_recovery.select_partial_roads(road_links: GeoDataFrame, road_nodes: GeoDataFrame, col_name: str, list_of_values: List[str]) Tuple[GeoDataFrame, GeoDataFrame][source]¶
Extract a subset of the road network based on specified road types.
Parameters¶
- road_links: gpd.GeoDataFrame
Geospatial data representing the links (edges) of the road network.
- road_nodes: gpd.GeoDataFrame
Geospatial data representing the nodes (vertices) of the road network.
- col_name: str
Column name in road_links that specifies the road type.
- list_of_values: List[str]
List of road types to filter and extract from the network.
Returns¶
- selected_links: gpd.GeoDataFrame
Filtered road links corresponding to the specified road types.
- selected_nodes: gpd.GeoDataFrame
Filtered road nodes connected to the selected links.
- nird.road_recovery.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_recovery.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_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_recovery.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_recovery.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_revised module¶
Road Network Flow Model - Functions
- nird.road_revised.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_revised.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.create_igraph_network(road_links: GeoDataFrame) Graph[source]¶
Create an undirected igraph network.
- nird.road_revised.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.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_revised.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_revised.edge_reclassification_func(road_links: DataFrame) DataFrame[source]¶
Reclassify network edges to “M, A_dual, A_single, B”.
- nird.road_revised.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.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.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.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.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.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.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_revised.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_revised.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_revised.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_revised.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_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