Refactor get_cartography return values and add to_folium_map functions

requirements.txt

@@ -5,8 +5,7 @@ networkx
 numpy
 osmnx
 pandas
-Pillow
 seaborn
 tqdm
-opencv-python
 shapely
+folium

setup.py

@@ -521,5 +521,6 @@ def run_stubgen(self):
         "numpy",
         "geopandas",
         "shapely",
+        "folium",
     ],
 )

src/dsf/__init__.py

@@ -21,4 +21,5 @@
     graph_from_gdfs,
     graph_to_gdfs,
     create_manhattan_cartography,
+    to_folium_map,
 )

src/dsf/dsf.hpp

@@ -6,7 +6,7 @@
 
 static constexpr uint8_t DSF_VERSION_MAJOR = 4;
 static constexpr uint8_t DSF_VERSION_MINOR = 7;
-static constexpr uint8_t DSF_VERSION_PATCH = 2;
+static constexpr uint8_t DSF_VERSION_PATCH = 3;
 
 static auto const DSF_VERSION =
     std::format("{}.{}.{}", DSF_VERSION_MAJOR, DSF_VERSION_MINOR, DSF_VERSION_PATCH);

src/dsf/python/cartography.py

@@ -7,6 +7,7 @@
 standardization of attributes.
 """
 
+import folium
 import geopandas as gpd
 import networkx as nx
 import numpy as np
@@ -21,8 +22,7 @@
     consolidate_intersections: bool | float = 10,
     dead_ends: bool = False,
     infer_speeds: bool = False,
-    return_type: str = "gdfs",
-) -> tuple | nx.DiGraph:
+) -> tuple[nx.DiGraph, gpd.GeoDataFrame, gpd.GeoDataFrame]:
     """
     Retrieves and processes cartography data for a specified place using OpenStreetMap data.
 
@@ -44,16 +44,14 @@
         infer_speeds (bool, optional): Whether to infer edge speeds based on road types. Defaults to False.
             If True, calls ox.routing.add_edge_speeds using np.nanmedian as aggregation function.
             Finally, the "maxspeed" attribute is replaced with the inferred "speed_kph", and the "travel_time" attribute is computed.
-        return_type (str, optional): Type of return value. Options are "gdfs" (GeoDataFrames) or
-            "graph" (NetworkX DiGraph). Defaults to "gdfs".
 
     Returns:
-        tuple | nx.DiGraph: If return_type is "gdfs", returns a tuple containing two GeoDataFrames:
+        tuple[nx.DiGraph, gpd.GeoDataFrame, gpd.GeoDataFrame]: Returns a tuple containing:
+            - NetworkX DiGraph with standardized attributes.
             - gdf_edges: GeoDataFrame with processed edge data, including columns like 'source',
               'target', 'nlanes', 'type', 'name', 'id', and 'geometry'.
             - gdf_nodes: GeoDataFrame with processed node data, including columns like 'id', 'type',
               and 'geometry'.
-            If return_type is "graph", returns the NetworkX DiGraph with standardized attributes.
     """
     if bbox is None and place_name is None:
         raise ValueError("Either place_name or bbox must be provided.")
@@ -223,32 +221,26 @@
         ):  # Check for NaN
             G.nodes[node]["type"] = "N/A"
 
-    # Return graph or GeoDataFrames based on return_type
-    if return_type == "graph":
-        return G
-    elif return_type == "gdfs":
-        # Convert back to MultiDiGraph temporarily for ox.graph_to_gdfs compatibility
-        gdf_nodes, gdf_edges = ox.graph_to_gdfs(nx.MultiDiGraph(G))
+    # Convert back to MultiDiGraph temporarily for ox.graph_to_gdfs compatibility
+    gdf_nodes, gdf_edges = ox.graph_to_gdfs(nx.MultiDiGraph(G))
 
-        # Reset index and drop unnecessary columns (id, source, target already exist from graph)
-        gdf_edges.reset_index(inplace=True)
-        # Move the "id" column to the beginning
-        id_col = gdf_edges.pop("id")
-        gdf_edges.insert(0, "id", id_col)
+    # Reset index and drop unnecessary columns (id, source, target already exist from graph)
+    gdf_edges.reset_index(inplace=True)
+    # Move the "id" column to the beginning
+    id_col = gdf_edges.pop("id")
+    gdf_edges.insert(0, "id", id_col)
 
-        # Ensure length is float
-        gdf_edges["length"] = gdf_edges["length"].astype(float)
+    # Ensure length is float
+    gdf_edges["length"] = gdf_edges["length"].astype(float)
 
-        gdf_edges.drop(columns=["u", "v", "key"], inplace=True, errors="ignore")
+    gdf_edges.drop(columns=["u", "v", "key"], inplace=True, errors="ignore")
 
-        # Reset index for nodes
-        gdf_nodes.reset_index(inplace=True)
-        gdf_nodes.drop(columns=["y", "x"], inplace=True, errors="ignore")
-        gdf_nodes.rename(columns={"osmid": "id"}, inplace=True)
+    # Reset index for nodes
+    gdf_nodes.reset_index(inplace=True)
+    gdf_nodes.drop(columns=["y", "x"], inplace=True, errors="ignore")
+    gdf_nodes.rename(columns={"osmid": "id"}, inplace=True)
 
-        return gdf_edges, gdf_nodes
-    else:
-        raise ValueError("Invalid return_type. Choose 'gdfs' or 'graph'.")
+    return G, gdf_edges, gdf_nodes
 
 
 def graph_from_gdfs(
@@ -460,6 +452,52 @@
     return gdf_edges, gdf_nodes
 
 
+def to_folium_map(
+    G: nx.DiGraph,
+    which: str = "edges",
+) -> folium.Map:
+    """
+    Converts a NetworkX DiGraph to a Folium map for visualization.
+    Args:
+        G (nx.DiGraph): The input DiGraph.
+        which (str): Specify whether to visualize 'edges', 'nodes', or 'both'. Defaults to 'edges'.
+    Returns:
+        folium.Map: The Folium map with the graph visualized.
+    """
+
+    # Compute mean latitude and longitude for centering the map
+    mean_lat = np.mean([data["geometry"].y for _, data in G.nodes(data=True)])
+    mean_lon = np.mean([data["geometry"].x for _, data in G.nodes(data=True)])
+    folium_map = folium.Map(location=[mean_lat, mean_lon], zoom_start=13)
+
+    if which in ("edges", "both"):
+        # Add edges to the map
+        for _, _, data in G.edges(data=True):
+            line = data.get("geometry")
+            if line:
+                folium.PolyLine(
+                    locations=[(point[1], point[0]) for point in line.coords],
+                    color="blue",
+                    weight=2,
+                    opacity=0.7,
+                    popup=f"Edge ID: {data.get('id')}",
+                ).add_to(folium_map)
+    if which in ("nodes", "both"):
+        # Add nodes to the map
+        for _, data in G.nodes(data=True):
+            folium.CircleMarker(
+                location=(data["geometry"].y, data["geometry"].x),
+                radius=5,
+                color="red",
+                fill=True,
+                fill_color="red",
+                fill_opacity=0.7,
+                popup=f"Node ID: {data.get('id')}",
+            ).add_to(folium_map)
+
+    return folium_map
+
+
 # if __name__ == "__main__":
 #     # Produce data for tests
 #     edges, nodes = get_cartography(

test/Test_cartography.py

@@ -4,11 +4,13 @@
 
 import pytest
 import networkx as nx
+import folium
 from dsf.python.cartography import (
     get_cartography,
     graph_to_gdfs,
     graph_from_gdfs,
     create_manhattan_cartography,
+    to_folium_map,
 )
 
 
@@ -17,10 +19,7 @@
     A simple consistency test to verify that converting from GeoDataFrames to graph and back
     yields the same GeoDataFrames.
     """
-    G_CART = get_cartography("Postua, Piedmont, Italy", return_type="graph")
-    edges_cart, nodes_cart = get_cartography(
-        "Postua, Piedmont, Italy", return_type="gdfs"
-    )
+    G_CART, edges_cart, nodes_cart = get_cartography("Postua, Piedmont, Italy")
 
     edges, nodes = graph_to_gdfs(G_CART)
 
@@ -221,5 +220,64 @@
         assert len(edges) == expected_edges
 
 
+class TestToFoliumMap:
+    """Tests for to_folium_map function."""
+
+    @pytest.fixture
+    def sample_graph(self):
+        """Create a sample graph for testing."""
+        edges, nodes = create_manhattan_cartography(n_x=3, n_y=3)
+        return graph_from_gdfs(edges, nodes)
+
+    def test_returns_folium_map(self, sample_graph):
+        """Test that the function returns a folium.Map object."""
+        result = to_folium_map(sample_graph)
+        assert isinstance(result, folium.Map)
+
+    def test_edges_only(self, sample_graph):
+        """Test visualization with edges only (default)."""
+        result = to_folium_map(sample_graph, which="edges")
+        assert isinstance(result, folium.Map)
+        # Check that the map has children (the edges)
+        assert len(result._children) > 0
+
+    def test_nodes_only(self, sample_graph):
+        """Test visualization with nodes only."""
+        result = to_folium_map(sample_graph, which="nodes")
+        assert isinstance(result, folium.Map)
+        assert len(result._children) > 0
+
+    def test_both_edges_and_nodes(self, sample_graph):
+        """Test visualization with both edges and nodes."""
+        result = to_folium_map(sample_graph, which="both")
+        assert isinstance(result, folium.Map)
+        # Should have more children than edges-only or nodes-only
+        edges_only = to_folium_map(sample_graph, which="edges")
+        nodes_only = to_folium_map(sample_graph, which="nodes")
+        # 'both' should have children from edges and nodes combined
+        # (minus the base tile layer which is common)
+        assert len(result._children) >= len(edges_only._children)
+        assert len(result._children) >= len(nodes_only._children)
+
+    def test_map_center_location(self, sample_graph):
+        """Test that the map is centered correctly."""
+        result = to_folium_map(sample_graph)
+        # The map should be centered around the mean of node coordinates
+        # For a Manhattan grid centered at (0, 0), the center should be near (0, 0)
+        location = result.location
+        assert location is not None
+        assert len(location) == 2
+        # Check that location is reasonable (near 0,0 for default manhattan grid)
+        assert -1 < location[0] < 1  # latitude
+        assert -1 < location[1] < 1  # longitude
+
+    def test_default_which_parameter(self, sample_graph):
+        """Test that default 'which' parameter is 'edges'."""
+        default_result = to_folium_map(sample_graph)
+        edges_result = to_folium_map(sample_graph, which="edges")
+        # Both should produce maps with the same number of children
+        assert len(default_result._children) == len(edges_result._children)
+
+
 if __name__ == "__main__":
     pytest.main([__file__, "-v"])