nird package¶

NIRD - National Infrastructure Resilience Demonstrator

Submodules¶

nird.constants module¶

Constants used in the network flow model

nird.road module¶

Road transport model functions

nird.road.cost_func(time: float, distance: float, voc: float) → float[source]¶

Calculate the time-equivalent cost for traveling

Parameters¶

time: float: The time of travel: hour
distance: float: The distance of travel: mile
voc:: Value of Cost: pound/km

Returns¶

float: Time-equivalent cost: hour

nird.road.create_igraph_network(node_name_to_index: Dict[str, int], road_links: GeoDataFrame, road_nodes: GeoDataFrame, initialSpeeds: Dict[str, float]) → Graph[source]¶

Network creation using igraph.

Parameters¶

node_name_to_index: dict: The relation between node’s name and node’s index in a network.
road_links: gpd.GeoDataFrame: Edges of a road network.
road_nodes: gpd.GeoDataFrame: Nodes of a road network.
initialSpeeds: dict: Free-flow speeds of different types of roads.

Returns¶

igraph.Graph: A road network.

nird.road.create_urban_mask(etisplus_urban_roads: GeoDataFrame) → GeoDataFrame[source]¶

Urban road classification

Parameters¶

etisplus_urban_roads: gpd.GeoDataFrame: ETIS road networks: https://ftp.demis.nl/outgoing/etisplus/datadeliverables/

Returns¶

gpd.GeoDataFrame: Polygons indicating urban areas of the Europe.

nird.road.filter_less_than_one(arr: ndarray[Any, dtype[float64]]) → ndarray[Any, dtype[float64]][source]¶: Convert values less than one to zero

nird.road.find_nearest_node(zones: GeoDataFrame, road_nodes: GeoDataFrame) → Dict[str, str][source]¶

Find nearest network node for each admin centroid

Parameters¶

zones: gpd.GeoDataFrame: Administrative population-weighted centroids.
road_nodes: gpd.GeoDataFrame: Nodes of road network.

Returns¶

dict: Convert from centroids to road nodes.

nird.road.initial_speed_func(road_type: str, form_of_road: str, free_flow_speed_dict: Dict[str, float]) → float | None[source]¶

Append free-flow speed to each road segment

Parameters¶

road_type: str: The column of road type.
form_of_road: str: The column of form of road.
free_flow_speed_dict: dict: Free-flow speeds of different combined road types.

Returns¶

float or None: The initial speed: miles/hour

nird.road.initialise_igraph_network(road_links: GeoDataFrame, initial_capacity_dict: Dict[str, int], initial_speed_dict: Dict[str, float], col_road_classification: str) → GeoDataFrame[source]¶

Road Network Initialisation

Parameters¶

road_links: gpd.GeoDataFrame: Edges of the road network.
initial_capacity_dict: dict: The capacity of different types of roads.
initial_speed_dict: dict: Free-flow speeds of different types of roads.
col_road_classification: str: The column of road classification.

Returns¶

gpd.GeoDataFrame: Road links with combined_label, flow counts, capacities, and flow rages.

nird.road.label_urban_roads(road_links: GeoDataFrame, urban_mask: GeoDataFrame) → GeoDataFrame[source]¶

Label road segments with “urban” attribute

Parameters¶

road_links: gpd.GeoDataFrame: Edges of the road network.
urban_mask: gpd.GeoDataFrame: Polygons indicating urban areas of the Europe.

Returns¶

gpd.GeoDataFrame: Edges of the road network with urban attributes.

nird.road.network_flow_model(network: Graph, road_links: GeoDataFrame, node_name_to_index: Dict[str, int], edge_index_to_name: Dict[int, str], list_of_origins: List[str], supply_dict: Dict[str, List[int]], destination_dict: Dict[str, List[str]], free_flow_speed_dict: Dict[str, float], flow_breakpoint_dict: Dict[str, int], min_speed_cap: Dict[str, float], urban_speed_cap: Dict[str, float], col_eid: str) → Tuple[Dict[str, float], Dict[str, int], Dict[str, int]][source]¶

Network flow model

Parameters¶

network: igraph.Graph: The road network.
road_links: gpd.GeoDataFrame: Edges of the road network.
node_name_to_index: dict: Convert from node name to node index of the road network.
edge_index_to_name: dict: Convert from edge index to edge name of the road network.
list_of_origins: list: Origins of the OD matrix.
supply_dict: dict: The counts of outbound trips from each origin of the OD matrix.
destination_dict: dict: List of destinations attached to each origin of the OD matrix.
free_flow_speed_dict: dict: Free-flow speeds of different combined road types, mile/hour.
min_speed_cap: dict: The restriction on the lowest flow rate, mile/hour.
urban_speed_cap: dict: The restriction on the maximum flow rate in urban areas, mile/hour.

Returns¶

acc_speed_dict: dict: The updated average flow rate on each road link.
acc_flow_dict: The accumulated flow amount on each road link.
acc_capacity_dict: The remaining capacity of each road link.

nird.road.od_interpret(od_matrix: DataFrame, zone_to_node: Dict[str, str], col_origin: str, col_destination: str, col_count: str) → Tuple[List[str], Dict[str, List[str]], Dict[str, List[float]]][source]¶

Generate a list of origins along with their associated destinations and outbound trips.

Parameters¶

od_matrix: pd.DataFrame: Trips between Origin-Destination pairs.
zone_to_node: dict: Dictionary for conversion from zone centroids to road nodes.
col_origin: str: Column of origins.
col_destination: str: Column of destinations.
col_count: str: Column that records the number of trips of each OD pair.

Returns¶

list_of_origins: list: Origins of the OD matrix
destination_dict: dict: Destinations attached to each origin.
supply_dict: dict: The number of outbound trips from each origin.

nird.road.select_partial_roads(road_links: GeoDataFrame, road_nodes: GeoDataFrame, col_name: str, list_of_values: List[str]) → Tuple[GeoDataFrame, GeoDataFrame][source]¶

Extract major roads

Parameters¶

road_links: gpd.GeoDataFrame: Edges of the road network.
road_nodes: gpd.GeoDataFrame: Junctions/nodes of the road network.
col_name: str: The name of column containing different road classifications.
list_of_values: list: The road types to be selected, e.g., [Motorways, A Roads, B Roads].

Returns¶

selected_links: gpd.GeoDataFrame: The selected types of road links
selected_nodes: gpd.GeoDataFrame: The selected types of road nodes.

nird.road.speed_flow_func(road_type: str, isUrban: int, vp: float, free_flow_speed_dict: Dict[str, float], flow_breakpoint_dict: Dict[str, int], min_speed_cap: Dict[str, float], urban_speed_cap: Dict[str, float]) → float | None[source]¶

Update the average flow rate of each road segment in terms of flow changes.

Parameters¶

road_type: str: The column of road type.
isUrban: int: 1: urban, 0: suburban
vp: float: The average daily flow, cars/day.
free_flow_speed_dict: dict: Free-flow speeds of different combined road types, mile/hour.
min_speed_cap: dict: The restriction on the lowest flow rate, mile/hour.
urban_speed_cap: dict: The restriction on the maximum flow rate in urban areas, mile/hour.

Returns¶

float OR None: The average flow rate, mile/hour.

nird.road.update_network_structure(network: Graph, length_dict: Dict[str, float], speed_dict: Dict[str, float], temp_edge_flow: DataFrame) → Tuple[Graph, Dict[int, str]][source]¶

Update the network structure (links and nodes)

Parameters¶

network: igraph.Graph: The road network.
length_dict: dict: Lengths of road links.
speed_dict:: The flow rates of road links.
temp_edge_flow: pd.DataFrame: A temporary file to record edge flows in the network flow model.

Returns¶

network: igraph.Graph: the updated network.
edge_index_to_name:: The updated edge_index_to_name dict.

nird.road.update_od_matrix(temp_flow_matrix: DataFrame, supply_dict: Dict[str, List[int]], destination_dict: Dict[str, List[str]]) → Tuple[List[str], Dict[str, List[int]], Dict[str, List[str]], float][source]¶

Drop the origin-destination pairs with no accessible link or unallocated trips

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: List of destinations attached with each origin.

Returns¶

new_list_of_origins: list new_supply_dict: dict new_destination: dict non_allocated_flow: float

nird.road.voc_func(speed: float) → float[source]¶

Calculate value of cost for travelling

Parameters¶

speed: float: The average flow rate: mile/hour

Returns¶

float: Unit cost: pound/km

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.ckdnearest(gdA: GeoDataFrame, gdB: GeoDataFrame) → GeoDataFrame[source]¶: Taken from https://gis.stackexchange.com/questions/222315/finding-nearest-point-in-other-geodataframe-using-geopandas

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

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