Metapath Construction for Heterogeneous GNNs¶

import geopandas as gpd
import osmnx as ox
import pandas as pd
import matplotlib.pyplot as plt
from matplotlib.lines import Line2D
from matplotlib.animation import FuncAnimation
import city2graph as c2g

# Download and project the street network to British National Grid (EPSG:27700) for metric distances
# Data should be obtained from Only in Soho, London
G = ox.graph_from_place(
    "Soho, London",
    network_type="all",
)

street_primary_nodes, street_primary_edges = c2g.nx_to_gdf(G)
street_primary_nodes = street_primary_nodes.to_crs(epsg=27700)
street_primary_edges = street_primary_edges.to_crs(epsg=27700)

amenity_tags = ["cafe", "restaurant", "pub", "bar", "museum", "theatre", "cinema"]
amenity_candidates = ox.features_from_place(
    "Soho, London",
    tags={"amenity": amenity_tags},
).to_crs(epsg=27700)

# Collapse complex Amenity geometries to points within the projected CRS
amenities = (
    amenity_candidates[["name", "amenity", "geometry"]]
    .copy()
    .explode(index_parts=False)
    .dropna(subset=["geometry"])
)
non_point_mask = ~amenities.geometry.geom_type.isin(["Point"])
amenities.loc[non_point_mask, "geometry"] = amenities.loc[non_point_mask, "geometry"].centroid
amenities = amenities.set_geometry("geometry")
amenities["name"] = amenities["name"].fillna(amenities["amenity"].str.title())
amenities = amenities[~amenities.geometry.is_empty]
amenities = amenities.drop_duplicates(subset="geometry").reset_index(drop=True)

street_dual_nodes, street_dual_edges = c2g.dual_graph((street_primary_nodes, street_primary_edges))

street_dual_nodes.geometry = street_dual_nodes.geometry.centroid

nodes_dict = {
    "amenity": amenities,
    "segment": street_dual_nodes
}

edges_dict = {
    ("segment", "connects_to", "segment"): street_dual_edges
}

# Connect amenities to segments
# bridge_nodes returns the node dictionary and a new dictionary of proximity edges
_, bridged_edges = c2g.bridge_nodes(
    nodes_dict=nodes_dict,
    proximity_method="knn",
    source_node_types=["amenity"],
    target_node_types=["segment"],
    k=1  # Connect to the single nearest segment
)

# Add the street network edges to the edges dictionary
edges_dict.update(bridged_edges)

# Materialize the metapath edges
result_nodes, result_edges = c2g.add_metapaths(nodes=nodes_dict,
                                               edges=edges_dict,
                                               sequence=sequence,
                                               new_relation_name="is_3_hop_nearby")

nodes_dict = {
    "amenity": amenities,
    "intersection": street_primary_nodes
}

edges_dict = {
    ("intersection", "connects_to", "intersection"): street_primary_edges
}

_, bridged_edges = c2g.bridge_nodes(
    nodes_dict=nodes_dict,
    proximity_method="knn",
    source_node_types=["amenity"],
    target_node_types=["intersection"],
    k=1
)

edges_dict.update(bridged_edges)

# set travel time in seconds assuming average walking speed of 4 km/h
walking_speed_kmh = 4
walking_speed_mps = walking_speed_kmh * 1000 / 3600  # convert km/h to m/s

# Add travel time attribute to edges
for edge_type, edge_gdf in edges_dict.items():
    edges_dict[edge_type]["travel_time_sec"] = edge_gdf.length / walking_speed_mps

# Connect amenities within 500 meters walking distance
weight_nodes, weight_edges = c2g.add_metapaths_by_weight(
    nodes=nodes_dict,
    edges=edges_dict,
    weight="travel_time_sec",
    threshold=60,
    new_relation_name="is_within_1_min",
    endpoint_type="amenity",
    directed=False
)