### Package Version Log Source: https://superblockify.city/usage Example of the initialization log output showing the current version. ```text 2026-03-23 16:19:48,495 | INFO | __init__.py:11 | superblockify version 1.0.2 ``` -------------------------------- ### Execution logs and statistics Source: https://superblockify.city/usage Example output logs generated during the partitioning process. ```text 2026-03-23 16:20:00,666 | INFO | tessellation.py:101 | Calculating edge cells for graph with 2309 edges. ``` ```text 2026-03-23 16:20:02,593 | INFO | tessellation.py:155 | Tessellated 1563 edge cells in 0:00:01. ``` ```text 2026-03-23 16:20:02,934 | INFO | ghsl.py:129 | Using the GHSL raster tiles for the bounding box (314241.6946056535, 6089247.837399746, 318635.011526757, 6093522.768369852). ``` ```text 2026-03-23 16:20:07,725 | WARNING | features.py:148 | CPLE_AppDefined in DeprecationWarning: 'Memory' driver is deprecated since GDAL 3.11. Use 'MEM' onwards. Further messages of this type will be suppressed. ``` ```text Distributing population over road cells: 0%| | 0/1563 [00:00 4911 self._mgr.column_setitem(icol, iindex, value, inplace_only=True) 4913 except (KeyError, TypeError, ValueError, LossySetitemError): 4914 # get_loc might raise a KeyError for missing labels (falling back 4915 # to (i)loc will do expansion of the index) 4916 # column_setitem will do validation that may raise TypeError, 4917 # ValueError, or LossySetitemError 4918 # set using a non-recursive method & reset the cache File ~/micromamba/envs/sb_env/lib/python3.12/site-packages/pandas/core/internals/managers.py:1518, in BlockManager.column_setitem(self, loc, idx, value, inplace_only) 1517 if inplace_only: -> 1518 col_mgr.setitem_inplace(idx, value) 1519 else: File ~/micromamba/envs/sb_env/lib/python3.12/site-packages/pandas/core/internals/managers.py:2220, in SingleBlockManager.setitem_inplace(self, indexer, value) 2217 if isinstance(arr, np.ndarray): 2218 # Note: checking for ndarray instead of np.dtype means we exclude 2219 # dt64/td64, which do their own validation. -> 2220 value = np_can_hold_element(arr.dtype, value) 2222 if isinstance(value, np.ndarray) and value.ndim == 1 and len(value) == 1: 2223 # NumPy 1.25 deprecation: https://github.com/numpy/numpy/pull/10615 File ~/micromamba/envs/sb_env/lib/python3.12/site-packages/pandas/core/dtypes/cast.py:1725, in np_can_hold_element(dtype, element) 1724 # Anything other than integer we cannot hold -> 1725 raise LossySetitemError 1726 if ( 1727 dtype.kind == "u" 1728 and isinstance(element, np.ndarray) 1729 and element.dtype.kind == "i" 1730 ): 1731 # see test_where_uint64 ``` -------------------------------- ### Initialize and Run BasePartitioner Source: https://superblockify.city/_modules/superblockify/partitioning/base Demonstrates how to instantiate the BasePartitioner using either a pre-loaded graph or a search string for automatic downloading. ```python >>> from superblockify.partitioning import BasePartitioner >>> import osmnx as ox >>> name, search_str = "Resistencia", "Resistencia, Chaco, Argentina" >>> graph = ox.graph_from_place(search_str, network_type="drive") >>> part = BasePartitioner(graph=graph, name=name) >>> part.run(make_plots=True) ``` ```python >>> from superblockify.partitioning import BasePartitioner >>> import osmnx as ox >>> part = BasePartitioner( ... name="Resistencia", search_str="Resistencia, Chaco, Argentina" ... ) >>> part.run(approach=True, make_plots=True, num_workers=6) ``` -------------------------------- ### Indentation error example Source: https://superblockify.city/guide/31_tessellation An example of an IndentationError encountered during function definition. ```python Cell In[7], line 1 def border_from_subgraph_shrinked(subgraph, buffer=1): ^ IndentationError: unexpected indent ``` -------------------------------- ### Initialize Partitioning Instance Source: https://superblockify.city/_modules/superblockify/partitioning/base Validates input parameters and initializes the graph structure, results directory, and instance variables. ```python if not isinstance(name, str) or name == "": raise ValueError("Name must be a non-empty string.") if not isinstance(city_name, str) or city_name == "": raise ValueError("City name must be a non-empty string.") if not ( search_str is None or (isinstance(search_str, str) and search_str != "") or ( isinstance(search_str, list) and all(isinstance(s, str) and s != "" for s in search_str) and search_str != [] ) ): raise KeyError( "Search_str must be a non-empty string or a list of non-empty strings." ) # First check weather a graph is found under GRAPH_DIR/city_name.graphml graph_path = join(Config.GRAPH_DIR, f"{city_name}.graphml") if exists(graph_path): self.graph = self.load_or_find_graph(city_name, search_str, reload_graph) elif search_str is not None: self.graph = self.load_or_find_graph(city_name, search_str) elif graph is not None: self.graph = graph else: raise ValueError("Either graph or search_str must be provided.") if max_nodes is not None and self.graph.number_of_nodes() > max_nodes: self.graph = reduce_graph(self.graph, max_nodes=max_nodes) remove_dead_ends_directed(self.graph) show_highway_stats(self.graph) show_graph_stats(self.graph) # Create results directory self.results_dir = join(Config.RESULTS_DIR, name) if not exists(self.results_dir): makedirs(self.results_dir) # Set Instance variables self.name: Final[str] = name self.city_name: str = city_name self.partitions: List[dict] | None = None self.components: List[dict] | None = None self.sparsified = None self.attribute_label: str | None = None self.attribute_dtype: type | None = None self.attr_value_minmax: tuple | None = None self.metric = Metric(unit) # Log initialization logger.info( "Initialized %s(%s) with %d nodes and %d edges.", self.name, self.__class__.__name__, len(self.graph.nodes), len(self.graph.edges), ) ``` -------------------------------- ### Initialize and Project Graph Source: https://superblockify.city/guide/31_tessellation Downloads a graph for a specific location and projects it into a coordinate system suitable for spatial analysis. ```python ky_graph = ox.graph_from_place("Kyiv, Ukraine", network_type="drive") ky_graph = ox.project_graph(ky_graph) ``` -------------------------------- ### Initialize ResidentialPartitioner Source: https://superblockify.city/usage Create a partitioner instance for a specific city using OpenStreetMap data. ```python part = sb.ResidentialPartitioner( name="Scheveningen_test", city_name="Scheveningen", search_str="Scheveningen, NL", unit="time", # "time", "distance", any other edge attribute, or None to count edges ) ``` -------------------------------- ### Create environment from YAML file Source: https://superblockify.city/_sources/installation.md Use this method if you have cloned the repository and wish to use the provided environment.yml file. ```bash conda env create --file environment.yml conda activate sb_env pip install superblockify ``` -------------------------------- ### Prepare edge polygons for tessellation Source: https://superblockify.city/_modules/superblockify/population/tessellation Merges edges with the same start and end nodes if their geometries are equal or reversed. ```python def get_edge_polygons(graph): """Prepare edge polygons for tessellation. This returns a GeoDataFrame where edges with same start and end node are merged, if their geometry is equal or a reversed version of the other. Parameters ---------- graph : networkx.MultiDiGraph The graph to tessellate. Returns ------- edges : geopandas.GeoDataFrame The edges with their polygons. """ edges = graph_to_gdfs( graph, nodes=False, edges=True, node_geometry=False, fill_edge_geometry=True ) # Merge two columns, when the geometry of one is equal or the reverse of the other # 1. Group edge ig (u, v) == (v, u) or (u, v) == (u, v) # Match by node pair (u, v) where u < v # get flat index of the first match edges["edge_id"] = edges.index.to_flat_index() # sort the edge_id tuples edges["node_pair"] = edges["edge_id"].apply(lambda x: tuple(sorted(x[:2]))) # 2. Aggregate if the geometry is equal or the reverse of the other # Merge columns if geometry_1 == geometry_2 or geometry_1 == geometry_2.reverse() # reverse geometry if x_start >= x_end (analogous to sorted node_pair) edges["geometry"] = edges["geometry"].apply( lambda x: x if x.coords[0] < x.coords[-1] else x.reverse() ) # 3. Group by node_pair and geometry, aggregate edge_id edges = ( edges.groupby(["node_pair", "geometry"]) .agg( { "edge_id": tuple, # tuple of edge indices } # tuple, because it is immutable and thus hashable for indexing ) .reset_index() ) edges.set_index("edge_id", inplace=True) edges.drop(columns=["node_pair"], inplace=True) return GeoDataFrame(edges, geometry="geometry", crs=graph.graph["crs"]) ``` -------------------------------- ### Reconstruct edge cells from Voronoi diagram Source: https://superblockify.city/_modules/superblockify/population/tessellation Example usage of the reconstruction process including plotting the resulting tessellation. ```python # Reconstruct edge cells from Voronoi point tessellation logger.debug("Reconstructing edge cells.") edge_cells = reconstruct_edge_cells( edge_voronoi_diagram, edge_indices, graph.graph["crs"] ) # Plot tessellation if show_plot: fig, axe = plt.subplots(figsize=(8, 8)) edge_cells.to_crs(graph.graph["crs"]).plot( ax=axe, cmap="tab20", edgecolor="white", alpha=0.5 ) plot_graph(graph, ax=axe, node_size=0, edge_color="black", edge_linewidth=0.5) axe.set_axis_off() axe.set_title("Edge tessellation") fig.tight_layout() logger.info( "Tessellated %d edge cells in %s.", len(edge_cells), timedelta(seconds=(datetime.now() - start_time).seconds), ) return edge_cells ``` -------------------------------- ### Import required libraries for population density analysis Source: https://superblockify.city/guide/32_edge_population Initializes the environment with necessary geospatial, network, and statistical libraries for processing road network population data. ```python import geopandas as gpd import matplotlib.pyplot as plt import numpy as np import osmnx as ox from networkx import ego_graph from matplotlib._color_data import CSS4_COLORS from rasterio.features import shapes from rasterstats import zonal_stats from scipy.spatial import Voronoi, voronoi_plot_2d from scipy.odr import ODR, Model, RealData from shapely import STRtree from shapely.geometry import shape from superblockify import get_edge_cells, get_ghsl, resample_load_window, \ add_edge_population, add_edge_cells, get_population_area ``` -------------------------------- ### Perform BFS ego graph traversal Source: https://superblockify.city/_modules/superblockify/partitioning/utils Traverses a graph starting from a representative node up to a maximum node count. ```python def _bfs_egograph(graph, rep_node_id, max_nodes): """Breadth-first search (BFS) to get the ego graph of the representative node. Parameters ---------- graph : networkx.classes.multidigraph.MultiDiGraph Graph to get the ego graph for. rep_node_id : int Node id of the seed node. max_nodes : int Maximum number of nodes in the ego graph. Yields ------ (node_id, level) : tuple Node id and level of the node in the BFS. Notes ----- Modified from :func:`networkx.algorithms.shortest_paths.unweighted._single_shortest_path_length`. """ seen = set([rep_node_id]) nextlevel = [rep_node_id] level = 0 n_total = len(graph.adj) # Yield the seed node yield (rep_node_id, level) # Yield the nodes in the BFS, as long as the sum of the next level and the seen # nodes are less than the maximum number of nodes while ( nextlevel and len(seen) + len([n_next for leaf in nextlevel for n_next in graph.adj[leaf]]) <= max_nodes ): level += 1 thislevel = nextlevel nextlevel = [] for leaf in thislevel: for n_next in graph.adj[leaf]: if n_next not in seen: seen.add(n_next) nextlevel.append(n_next) # Yield the node id and level yield (n_next, level) if len(seen) == n_total: return # pragma: no cover ``` -------------------------------- ### Importing necessary libraries Source: https://superblockify.city/guide/31_tessellation Initial imports required for geospatial analysis and superblockify configuration. ```python import geopandas as gpd import matplotlib.pyplot as plt import osmnx as ox import shapely as shp import superblockify as sb from superblockify.config import logger, Config ``` -------------------------------- ### get_edge_polygons(graph) Source: https://superblockify.city/_modules/superblockify/population/tessellation Prepares edge polygons for tessellation by merging edges with the same start and end nodes if their geometry is equal or reversed. ```APIDOC ## get_edge_polygons(graph) ### Description Prepares edge polygons for tessellation. This returns a GeoDataFrame where edges with same start and end node are merged, if their geometry is equal or a reversed version of the other. ### Parameters - **graph** (networkx.MultiDiGraph) - Required - The graph to tessellate. ### Returns - **edges** (geopandas.GeoDataFrame) - The edges with their polygons. ``` -------------------------------- ### Initialize Logging and Configuration Source: https://superblockify.city/_modules/superblockify/config Configures the logger using an external file and sets the logging level for the numba library. ```python import logging.config from os import getcwd from os.path import join, dirname, exists from ruamel.yaml import YAML # Logging configuration using the logging.cfg file logging.config.fileConfig(join(dirname(__file__), "logging.cfg")) # Get the logger for this module numba_logger = logging.getLogger("numba") numba_logger.setLevel(logging.WARNING) logger = logging.getLogger("superblockify") ``` -------------------------------- ### reduce_graph(graph, max_nodes) Source: https://superblockify.city/_modules/superblockify/partitioning/utils Reduces a graph to a maximum number of nodes by extracting an ego graph starting from a representative central node. ```APIDOC ## reduce_graph(graph, max_nodes) ### Description Reduces a graph to have at most `max_nodes` nodes. It uses a breadth-first search starting from a representative node to create a subgraph. ### Parameters - **graph** (NetworkX Graph) - Required - The input graph to be reduced. - **max_nodes** (int) - Required - The maximum number of nodes allowed in the resulting graph. ### Returns - **graph** (NetworkX Graph) - The reduced subgraph with updated population and area statistics. ``` -------------------------------- ### Running the ResidentialPartitioner Source: https://superblockify.city/guide/31_tessellation Configures and executes the partitioner for a specific city to generate street tessellation and metrics. ```python CITY_NAME, SEARCH_STR = Config.PLACES_SMALL[2] logger.info( "Running partitioner for %s with search string %s.", CITY_NAME, SEARCH_STR ) part = sb.ResidentialPartitioner( name=CITY_NAME + "_main", city_name=CITY_NAME, search_str=SEARCH_STR ) part.run(calculate_metrics=False, make_plots=True) ``` -------------------------------- ### Process partitions and visualize Source: https://superblockify.city/guide/10_dist_calc_restricted Iterate through partitions to assign subgraphs and node colors, then visualize the full graph. ```python for name, part in partitions.items(): if "subgraph" not in part: # subgraph for all edges from or to nodes in partition part["subgraph"] = G.edge_subgraph( [(u, v) for u, v in G.edges if u in part["nodes"] or v in part["nodes"]] ) part["nodelist"] = part["subgraph"].nodes for node in part["nodes"]: G.nodes[node]["partition"] = part["color"] nx.draw(G, with_labels=True, node_color=[G.nodes[n]["partition"] for n in G.nodes], font_color="white", pos=nx.kamada_kawai_layout(G), ax=plt.figure(figsize=(8,3)).gca(), ) ``` -------------------------------- ### Get edge population for graph Source: https://superblockify.city/_modules/superblockify/population/approximation Calculates population and area for graph edges using GHSL data, utilizing parallel processing for efficiency. ```python def get_edge_population(graph, batch_size=10000, **tess_kwargs): """Get edge population for the graph. Calculates the population and area of the edge. First tessellates the edges and then determines the population with GHSL data. The population distribution process is parallelized with multiprocessing in batches of edges. ``` -------------------------------- ### Get partition nodes Source: https://superblockify.city/_modules/superblockify/partitioning/base Returns a list of dictionaries containing partition names, subgraphs, and node sets. Excludes nodes in the sparsified graph and ignored components. ```python def get_partition_nodes(self): """Get the nodes of the partitioned graph. Returns list of dict with name of partition and list of nodes in partition. If partitions were split up into components with `make_subgraphs_from_attribute` and `split_disconnected` is set to True, the nodes of the components are returned. Per default, nodes are considered to be inside a partition if they are in the subgraph of the partition and not in the sparsified subgraph. Also, `ignored` components are left out. Nodes inside the sparsified graph are never inside a partition. Method can be overwritten by child classes to change the definition of which nodes to consider to be inside a partition. Returns ------- list of dict List of dict with `name` of partition, `subgraph` of partition and set of `nodes` in partition. Raises ------ AssertionError If BasePartitioner has not been run yet (the partitions are not defined). """ self.check_has_been_run() # List of partitions /unignored components # Only take `name`, `subgraph` and `representative_node_id` from the components partitions = [ { "name": part["name"], "subgraph": part["subgraph"], "rep_node": part["representative_node_id"], } for part in self.get_ltns() # if there is the key "ignore" and it is True, ignore the partition if not (part.get("ignore") is True) ] # Add a list of nodes "inside" each partition # - nodes that are not shared with the sparsified graph are considered to be # inside the partition # - from these the distances are calculated # - the nodes not in any partitions are considered as the unpartitioned nodes for part in partitions: part["nodes"] = { node for node in part["subgraph"].nodes() if part["subgraph"].degree(node) >= 1 and node not in self.sparsified.nodes } return partitions ``` -------------------------------- ### Initialize Graph for Population Analysis Source: https://superblockify.city/guide/32_edge_population Load and project a graph from a specific location to prepare for population attribute calculation. ```python kd_graph = ox.graph_from_place("11th District, Kandahar, Kandahar Province, " "Afghanistan", network_type="drive") kd_graph = ox.project_graph(kd_graph) ``` -------------------------------- ### Get GHSL Raster URLs Source: https://superblockify.city/_modules/superblockify/population/ghsl Retrieves the URLs for GHSL population raster tiles covering a specific bounding box. Raises a ValueError if the bounding box spans more than two tiles in either dimension. ```python def get_ghsl_urls(bbox_moll): """Get the URLs of the GHSL population raster tiles that contain the boundary. Parameters ---------- bbox_moll : list Boundary of the place in Mollweide projection. [minx, miny, maxx, maxy] Returns ------- list URLs of the GHSL population raster tiles that contain the boundary. Raises ------ ValueError If the bounding box spans more than two tiles in each dimension. ValueError If the bounding box spans empty tile. Notes ----- Bounding boxes spanning areas larger than two tiles in each dimension are not supported. Use the whole raster instead. """ corners = [ (bbox_moll[0], bbox_moll[1]), # Lower left (bbox_moll[0], bbox_moll[3]), # Upper left (bbox_moll[2], bbox_moll[1]), # Lower right (bbox_moll[2], bbox_moll[3]), # Upper right ] tiles = {row_col(lat, long) for long, lat in corners} logger.debug("The bounding box spans the tiles %s.", tiles) if len(tiles) > 1: # that the difference between the min and max row/col is maximally 1 if max(tiles, key=lambda x: x[0])[0] - min(tiles, key=lambda x: x[0])[0] > 1: raise ValueError( "The bounding box spans more than two tiles in the row dimension." ) if max(tiles, key=lambda x: x[1])[1] - min(tiles, key=lambda x: x[1])[1] > 1: raise ValueError( "The bounding box spans more than two tiles in the column dimension." ) # Convert to urls tiles = { f"https://jeodpp.jrc.ec.europa.eu/ftp/jrc-opendata/GHSL" f"/GHS_POP_GLOBE_R2023A/GHS_POP_E2025_GLOBE_R2023A_54009_100/V1-0/tiles/" f"GHS_POP_E2025_GLOBE_R2023A_54009_100_V1_0_R{row}_C{col}.zip" for row, col in tiles } return list(tiles) ``` -------------------------------- ### Import superblockify Source: https://superblockify.city/usage Import the package using the standard sb alias. ```python import superblockify as sb ``` -------------------------------- ### Configure library settings Source: https://superblockify.city/usage Set global configuration options such as logging levels and raster file paths. ```python sb.config.set_log_level("DEBUG") ``` ```python sb.config.Config.FULL_RASTER = None ``` -------------------------------- ### Import necessary libraries Source: https://superblockify.city/guide/10_dist_calc_restricted Initial imports for graph manipulation, numerical operations, and visualization. ```python from itertools import combinations import networkx as nx import numpy as np import pandas as pd from matplotlib import pyplot as plt from scipy.sparse import csr_matrix from scipy.sparse.csgraph import dijkstra ``` -------------------------------- ### load Source: https://superblockify.city/_modules/superblockify/partitioning/base Loads a partitioner object from the filesystem based on the provided name. ```APIDOC ## load(name) ### Description Loads a partitioner from the filesystem. The name corresponds to the folder where the partitioner is saved. ### Parameters - **name** (str) - Required - Name of the partitioner folder. ### Returns - **partitioner** (BasePartitioner) - The loaded partitioner instance. ### Raises - **FileNotFoundError** - If the partitioner file cannot be found. ``` -------------------------------- ### Run Partitioning Pipeline Source: https://superblockify.city/_modules/superblockify/partitioning/base Executes the partitioning process, including graph partitioning, representative node setting, and validation. ```python @final def run( self, calculate_metrics=True, make_plots=False, replace_max_speeds=False, **part_kwargs, ): """Run partitioning. Parameters ---------- calculate_metrics : str or bool, optional If True, calculate metrics for the partitioning. If False, don't. Save them to self.results_dir/metrics. Default is True. make_plots : bool, optional If True, make plots of the partitioning and save them to self.results_dir/figures. Default is False. replace_max_speeds : bool, optional If True and :attr:`self.metric.unit` is "time", calculate the quickest paths in the restricted graph with the max speeds :attr:`V_MAX_LTN` and :attr:`V_MAX_SPARSE` set in :mod:`superblockify.config`. Default is False. **part_kwargs Passed to :meth:`self.partition_graph`. Depends on the abstract method :meth:`self.partition_graph` of the subclass. Warnings -------- If :attr:`self.metric.unit` is not "time", replace_max_speeds is ignored. """ # Warn if replace_max_speeds is not None when unit is not "time" if self.metric.unit != "time" and replace_max_speeds is not None: logger.warning("replace_max_speeds is ignored when unit is not 'time'.") replace_max_speeds = None self.partition_graph(make_plots=make_plots, **part_kwargs) # Set representative nodes set_representative_nodes(self.get_ltns()) # Calculate travel times for partitioned graph add_edge_travel_times_restricted(self.graph, self.sparsified) # Check that the partitions and sparsified graph satisfy the requirements is_valid_partitioning(self) ``` -------------------------------- ### Create and activate environment via Conda Source: https://superblockify.city/_sources/installation.md Use this command to create a new environment named sb_env with the necessary dependencies from the conda-forge channel. ```bash conda create -n sb_env -c conda-forge superblockify conda activate sb_env ``` -------------------------------- ### Visualize graph with partition colors Source: https://superblockify.city/guide/10_dist_calc_restricted Draws a graph using the Kamada-Kawai layout, coloring nodes based on their partition attribute. ```python nx.draw(G_2, with_labels=True, node_color=[G_2.nodes[n]["partition"] for n in G_2 .nodes], font_color="white", pos=nx.kamada_kawai_layout(G_2), ax=plt.figure(figsize=(8, 5)).gca(), ) ``` -------------------------------- ### Creating a planar graph for centrality analysis Source: https://superblockify.city/guide/20_betweenness_centrality Constructs a MultiDiGraph with specific weights and partition assignments to demonstrate centrality calculations. ```python # Create planar graph, similar to a street network G = nx.MultiDiGraph(nx.Graph( [ (10, 11, {"weight": 1}), (11, 12, {"weight": 1}), (12, 13, {"weight": 1}), (13, 0, {"weight": 1.5}), (13, 14, {"weight": 1}), (14, 0, {"weight": 1}), (0, 10, {"weight": 1}), (0, 1, {"weight": 1}), (10, 1, {"weight": 1}), (1, 2, {"weight": 1}), (2, 3, {"weight": 1}), (3, 4, {"weight": 1.5}), (4, 5, {"weight": 1}), (5, 9, {"weight": 1}), (5, 6, {"weight": 1}), (7, 2, {"weight": 1}), (8, 7, {"weight": 0.5}), (7, 1, {"weight": 1}), (8, 9, {"weight": 0.7}), (6, 9, {"weight": 1}), (8, 4, {"weight": 1}), (9, 1, {"weight": 1}), (0, 18, {"weight": 0.4}), (18, 2, {"weight": 0.4}), (6, 15, {"weight": 0.8}), (15, 16, {"weight": 1}), (16, 17, {"weight": 1}), (17, 6, {"weight": 1}), ] )) G.add_node(19) # isolated node # Delete directed edges (1, 9), (6, 17), (10, 1) G.remove_edges_from([(1, 9), (6, 17), (10, 1)]) # Add longer edge 0 -> 13 G.add_edge(0, 13, weight=G[0][13][0]["weight"] * 2) n_sparse = [0, 1, 2, 3, 4, 5, 6, 19] partitions = { "sparsified": {"nodes": n_sparse, "color": "black", "subgraph": G.subgraph(n_sparse)}, "G_r": {"nodes": [7, 8, 9], "color": "crimson"}, "G_g": {"nodes": [10, 11, 12, 13, 14], "color": "mediumseagreen"}, "G_b": {"nodes": [15, 16, 17], "color": "dodgerblue"}, "G_o": {"nodes": [18], "color": "darkorange"}, } for name, part in partitions.items(): if "subgraph" not in part: # subgraph for all edges from or to nodes in partition part["subgraph"] = G.edge_subgraph( # [(u, v) for u, v in G.edges if u in part["nodes"] or v in part["nodes"]] [e for e in G.edges if e[0] in part["nodes"] or e[1] in part["nodes"]] ) part["nodelist"] = part["subgraph"].nodes for node in part["nodes"]: G.nodes[node]["partition"] = part["color"] nx.draw(G, with_labels=True, node_color=[G.nodes[n]["partition"] for n in G .nodes], font_color="white", pos=nx.kamada_kawai_layout(G), ax=plt.figure(figsize=(8, 5)).gca(), connectionstyle="arc3,rad=0.1", ) ``` -------------------------------- ### Retrieve Superblock list Source: https://superblockify.city/_modules/superblockify/partitioning/base Returns the list of components if available, otherwise returns the partitions. ```python def get_ltns(self): """Get Superblock list. Returns ------- list Reference to self.components, if it is used, otherwise self.partitions """ return self.components if self.components else self.partitions ``` -------------------------------- ### Save and Load Metrics Source: https://superblockify.city/_modules/superblockify/metrics/metric Methods for persisting metric objects to disk using pickle and loading them back. ```python def save(self, folder, name, dismiss_distance_matrix=False): """Save the metric to a file. Will be saved as a pickle file at folder/name.metrics. Parameters ---------- folder : str The folder to save the metric to. name : str The name of the file to save the metric to. dismiss_distance_matrix : bool, optional If True, the distance matrix will not be saved. Default is False. """ metrics_path = join(folder, name + ".metrics") # Check if metrics already exist if exists(metrics_path): logger.debug("Metrics already exist, overwriting %s", metrics_path) else: logger.debug("Saving metrics to %s", metrics_path) if dismiss_distance_matrix: # Get distance_matrix and predescessor_matrix reference distance_matrix = self.distance_matrix predecessor_matrix = self.predecessor_matrix # Set distance_matrix and predescessor_matrix to None self.distance_matrix = None self.predecessor_matrix = None with open(metrics_path, "wb") as file: pickle.dump(self, file) if dismiss_distance_matrix: # Set distance_matrix and predescessor_matrix back to original self.distance_matrix = distance_matrix self.predecessor_matrix = predecessor_matrix ``` ```python @classmethod def load(cls, name): """Load a partitioning from a file. Parameters ---------- path : str The path to the file to load the partitioning from. Returns ------- partitioning : Partitioning The loaded partitioning. """ metrics_path = join(Config.RESULTS_DIR, name, name + ".metrics") logger.debug("Loading metrics from %s", metrics_path) with open(metrics_path, "rb") as file: metrics = pickle.load(file) return metrics ``` -------------------------------- ### Metric Calculation Methods Source: https://superblockify.city/_modules/superblockify/partitioning/base Methods for calculating network metrics before and after partitioning. ```python def calculate_metrics( self, make_plots=False, replace_max_speeds=True, ): """Calculate metrics for the partitioning. Calculates the metrics for the partitioning and writes them to the metrics dictionary. It includes the network metrics for the partitioned graph. There are different network measures - :math:`d_E(i, j)`: Euclidean - :math:`d_S(i, j)`: Shortest path on full graph - :math:`d_N(i, j)`: Shortest path with a ban through Superblocks Parameters ---------- make_plots : bool, optional If True show visualization graphs of the approach. If False, only print into console. Default is False. replace_max_speeds : bool, optional If True and unit is "time", replace max speeds set in :mod:`superblockify.config`. Default is True. """ # Log calculating metrics logger.info("Calculating metrics for %s", self.name) self.metric.calculate_all( partitioner=self, replace_max_speeds=replace_max_speeds, make_plots=make_plots, ) logger.debug("Metrics for %s: %s", self.name, self.metric) ``` ```python def calculate_metrics_before(self, make_plots=False, betweenness_range=None): """Calculate metrics for the graph before partitioning. Calculates the metrics that might be used by a partitioning approach. It includes the network shortest paths :math:`d_S(i, j)` on the full graph and betweenness centralities. Parameters ---------- make_plots : bool, optional If True show visualization graphs of the approach. If False, only print into console. Default is False. betweenness_range : int, optional The range to use for calculating the betweenness centrality, by default None, which uses the whole graph. Its unit is meters. If it is not None, two types of betweenness centralities are calculated. The whole one is always needed for other statistics. """ # Log calculating metrics logger.info("Calculating metrics for %s before partitioning", self.name) self.metric.calculate_before( partitioner=self, make_plots=make_plots, betweenness_range=betweenness_range ) logger.debug("Metrics for %s: %s", self.name, self.metric) ``` -------------------------------- ### load_graph_from_place(save_as, search_string, add_population=False, only_cache=False, max_nodes=Config.MAX_NODES, **gfp_kwargs) Source: https://superblockify.city/_modules/superblockify/utils Loads a graph from a specified place using OpenStreetMap data and saves it to a file. ```APIDOC ## load_graph_from_place ### Description Loads a graph from a place name or OSM relation ID, performs preprocessing, and saves it. It supports filtering by node count and passing additional arguments to the underlying OSMnx graph loader. ### Parameters - **save_as** (str) - Required - The name of the file to save the graph to. - **search_string** (str or list of str) - Required - The place name or OSM relation ID (e.g., "R1234567"). - **add_population** (bool) - Optional - Whether to add population data to the graph. Default is False. - **only_cache** (bool) - Optional - Whether to only load the graph from cache. Default is False. - **max_nodes** (int) - Optional - Maximum number of nodes in the graph. Default is Config.MAX_NODES. - **gfp_kwargs** (dict) - Optional - Keyword arguments to pass to osmnx.graph_from_place. ### Returns - **graph** (networkx.MultiDiGraph or None) - The loaded graph, or None if only_cache is True. ``` -------------------------------- ### Calculate shortest paths with partition restrictions Source: https://superblockify.city/_modules/superblockify/metrics/distances This snippet demonstrates the preparation of node indices and the construction of semipermeable sparse matrices for distance calculations. ```python # node_order indices # filtered indices: sparse/partition n_sparse_indices = [node_order.index(n) for n in partitions["sparsified"]["nodes"]] part_name_order = [name for name in partitions.keys() if name != "sparsified"] n_partition_indices_separate = [ [node_order.index(n) for n in partitions[name]["nodes"]] for name in part_name_order ] n_partition_indices = [n for part in n_partition_indices_separate for n in part] # Semipermeable graphs logger.debug("Preparing semipermeable graphs.") # Construct Compressed Sparse Row matrix # https://docs.scipy.org/doc/scipy/reference/generated/scipy.sparse.coo_matrix.html g_leaving = to_scipy_sparse_array( graph, nodelist=node_order, weight=weight, format="coo" ) data, row, col = g_leaving.data, g_leaving.row, g_leaving.col mask_to = np.logical_and( np.isin(row, n_sparse_indices), np.isin(col, n_partition_indices) ) mask_from = np.logical_and( np.isin(row, n_partition_indices), np.isin(col, n_sparse_indices) ) # mask the edges inside each partition and inside the sparse graph mask_intra = np.logical_and( np.isin(row, n_sparse_indices), np.isin(col, n_sparse_indices) ) for n_ind in n_partition_indices_separate: mask_intra = np.logical_or( mask_intra, np.logical_and(np.isin(row, n_ind), np.isin(col, n_ind)) ) mask_to = np.logical_or(mask_to, mask_intra) mask_from = np.logical_or(mask_from, mask_intra) g_leaving = csr_matrix( (data[mask_to], (row[mask_to], col[mask_to])), shape=(len(node_order), len(node_order)), ) g_entering = csr_matrix( (data[mask_from], (row[mask_from], col[mask_from])), shape=(len(node_order), len(node_order)), ) ``` -------------------------------- ### Project Bounding Box and Retrieve URLs Source: https://superblockify.city/guide/30_population_density Projects a GeoDataFrame to World Mollweide and retrieves the corresponding GHSL tile URLs. ```python # Project to World Mollweide to get the tile urls bogota_gdf = bogota_gdf.to_crs("World Mollweide") bogota_bbox = list(bogota_gdf.bounds.values[0]) bogota_url = get_GHSL_urls(bogota_bbox) ``` -------------------------------- ### Compare Components and Partitions Source: https://superblockify.city/_modules/superblockify/utils Compares two lists of dictionaries, supporting isomorphism checks for networkx graphs. ```python def compare_components_and_partitions(list1, list2): """Compare two lists of dictionaries. The lists are equal if they have the same length and the dictionaries in the lists are equal. If a value of a dictionary is a networkx.Graph, it compares the graphs with networkx.graph_equal(). Parameters ---------- list1 : list of dict The first list of dictionaries to compare. list2 : list of dict The second list of dictionaries to compare. Returns ------- bool True if the lists are equal, False otherwise. """ if len(list1) != len(list2): return False for element1, element2 in zip(list1, list2): if element1.keys() != element2.keys(): return False for key in element1.keys(): if all(isinstance(x, Graph) for x in [element1[key], element2[key]]): # Check if Graphs are isomorphic, as the attributes might differ if not is_isomorphic(element1[key], element2[key]): return False elif element1[key] != element2[key]: return False return True ``` -------------------------------- ### Calculate All Partitioning Metrics Source: https://superblockify.city/_modules/superblockify/metrics/metric Computes all remaining metrics for a specific partitioning. ```python def calculate_all( self, partitioner, replace_max_speeds=True, make_plots=False, ): """Calculate all metrics for the partitioning. If :meth:`calculate_before` has been called before partitioning, only the remaining metrics are calculated. ``` -------------------------------- ### Configure Jupyter kernel Source: https://superblockify.city/_sources/installation.md Run these commands to register the sb_env environment as a kernel available within Jupyter notebooks. ```bash pip install --user ipykernel python -m ipykernel install --user --name=sb_env ``` -------------------------------- ### save(save_graph_copy=False, dismiss_distance_matrix=False, key_figures=False) Source: https://superblockify.city/_modules/superblockify/partitioning/base Pickles the partitioner and saves it to a file, with options to save graph copies or key figures. ```APIDOC ## save ### Description Pickles the partitioner and saves it to a file. Metric objects are saved in a separate file. ### Parameters - **save_graph_copy** (bool) - Optional - If True, save a copy of the graph modified by the partitioner. - **dismiss_distance_matrix** (bool) - Optional - If True, dismiss the distance matrices in the metric object before saving. - **key_figures** (bool) - Optional - If True, save key figures to file. ```