Merge pull request #306 from Deltares/feature/178-enable-osmnetworkwr…

…apper-creation-without-providing-a-file

Feature/178 enable osmnetworkwrapper creation without providing a file

ra2ce/network/network_wrappers/osm_network_wrapper/osm_network_wrapper.py

@@ -15,17 +15,19 @@
     along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 
+from __future__ import annotations
 import logging
 from pathlib import Path
-from typing import Any, Union
+from typing import Any
 
 import networkx as nx
 import osmnx
 import pandas as pd
 from geopandas import GeoDataFrame
 from networkx import MultiDiGraph, MultiGraph
-from shapely.geometry import LineString
 from shapely.geometry.base import BaseGeometry
+from ra2ce.network.network_config_data.enums.network_type_enum import NetworkTypeEnum
+from ra2ce.network.network_config_data.enums.road_type_enum import RoadTypeEnum
 
 import ra2ce.network.networks_utils as nut
 from ra2ce.network.exporters.json_exporter import JsonExporter
@@ -39,31 +41,51 @@
 
 
 class OsmNetworkWrapper(NetworkWrapperProtocol):
+
+    polygon_graph: MultiDiGraph
+    network_type: NetworkTypeEnum
+    road_types: list[RoadTypeEnum]
+
     def __init__(self, config_data: NetworkConfigData) -> None:
         self.output_graph_dir = config_data.output_graph_dir
         self.graph_crs = config_data.crs
 
         # Network
-        self.network_type = config_data.network.network_type.config_value
-        self.road_types = list(
-            _enum.config_value for _enum in config_data.network.road_types
-        )
-        self.polygon_path = config_data.network.polygon
+        self.network_type = config_data.network.network_type
+        self.road_types = config_data.network.road_types
+        self.polygon_graph = self._get_clean_graph_from_osm(config_data.network.polygon)
         self.is_directed = config_data.network.directed
 
-    def get_network(self) -> tuple[MultiGraph, GeoDataFrame]:
+    @classmethod
+    def with_polygon(
+        cls, config_data: NetworkConfigData, polygon: BaseGeometry
+    ) -> OsmNetworkWrapper:
         """
-        Gets an indirected graph
+        Gets an `OsmNetworkWrapper` with the given `polygon` transformed into a
+        clean graph as the `polygon_graph` property.
+
+        Args:
+            config_data (NetworkConfigData): Basic configuration data which contain information required the different methods of this wrapper.
+            polygon (BaseGeometry): Base polygon from which to generate the graph.
 
         Returns:
-            tuple[MultiGraph, GeoDataFrame]: _description_
+            OsmNetworkWrapper: Wrapper with valid `polygon_graph` property.
         """
+        _wrapper = cls(config_data)
+        _clean_graph = _wrapper._download_clean_graph_from_osm(
+            polygon=polygon,
+            network_type=_wrapper.network_type,
+            road_types=_wrapper.road_types,
+        )
+        _wrapper.polygon_graph = _clean_graph
+        return _wrapper
+
+    def get_network(self) -> tuple[MultiGraph, GeoDataFrame]:
         logging.info("Start downloading a network from OSM.")
-        graph_complex = self.get_clean_graph_from_osm()
 
         # Create 'graph_simple'
         graph_simple, graph_complex, link_tables = nut.create_simplified_graph(
-            graph_complex
+            self.polygon_graph
         )
 
         # Create 'edges_complex', convert complex graph to geodataframe
@@ -121,25 +143,25 @@ def _export_linking_tables(self, linking_tables: tuple[Any]) -> None:
             self.output_graph_dir.joinpath("complex_to_simple.json"), linking_tables[1]
         )
 
-    def get_clean_graph_from_osm(self) -> MultiDiGraph:
+    def _get_clean_graph_from_osm(self, polygon_path: Path) -> MultiDiGraph | None:
         """
-        Creates a network from a polygon by by downloading via the OSM API in its extent.
+        Creates a network from a polygon by downloading via the OSM API in its extent.
 
-        Raises:
-            FileNotFoundError: When no valid polygon file is provided.
+        Args:
+            polygon_path (Path): Path where the polygon file can be found.
 
         Returns:
             MultiDiGraph: Complex (clean) graph after download from OSM, for use in the direct analyses and input to derive simplified network.
         """
         # It can only read in one geojson
-        if not isinstance(self.polygon_path, Path):
-            raise ValueError("No valid value provided for polygon file.")
-        if not self.polygon_path.is_file():
-            raise FileNotFoundError(
-                "No polygon_file file found at {}.".format(self.polygon_path)
-            )
-
-        _normalized_polygon = nut.get_normalized_geojson_polygon(self.polygon_path)
+        if not isinstance(polygon_path, Path):
+            logging.warning("No valid value provided for polygon file.")
+            return None
+        elif not polygon_path.is_file():
+            logging.error("No polygon_file file found at {}.".format(polygon_path))
+            return None
+
+        _normalized_polygon = nut.get_normalized_geojson_polygon(polygon_path)
         _complex_graph = self._download_clean_graph_from_osm(
             polygon=_normalized_polygon,
             network_type=self.network_type,
@@ -148,31 +170,40 @@ def get_clean_graph_from_osm(self) -> MultiDiGraph:
         return _complex_graph
 
     def _download_clean_graph_from_osm(
-        self, polygon: BaseGeometry, road_types: list[str], network_type: str
+        self,
+        polygon: BaseGeometry,
+        road_types: list[RoadTypeEnum],
+        network_type: NetworkTypeEnum,
     ) -> MultiDiGraph:
         _available_road_types = road_types and any(road_types)
+        _road_types_as_str = (
+            list(map(lambda x: x.config_value, road_types))
+            if _available_road_types
+            else []
+        )
+
         if not _available_road_types and not network_type:
             raise ValueError("Either of the link_type or network_type should be known")
         elif not _available_road_types:
             # The user specified only the network type.
             _complex_graph = osmnx.graph_from_polygon(
                 polygon=polygon,
-                network_type=network_type,
+                network_type=network_type.config_value,
                 simplify=False,
                 retain_all=True,
             )
         elif not network_type:
             # The user specified only the road types.
-            cf = f'["highway"~"{"|".join(road_types)}"]'
+            cf = f'["highway"~"{"|".join(_road_types_as_str)}"]'
             _complex_graph = osmnx.graph_from_polygon(
                 polygon=polygon, custom_filter=cf, simplify=False, retain_all=True
             )
         else:
             # _available_road_types and network_type
-            cf = f'["highway"~"{"|".join(road_types)}"]'
+            cf = f'["highway"~"{"|".join(_road_types_as_str)}"]'
             _complex_graph = osmnx.graph_from_polygon(
                 polygon=polygon,
-                network_type=network_type,
+                network_type=network_type.config_value,
                 custom_filter=cf,
                 simplify=False,
                 retain_all=True,

tests/network/network_wrappers/test_osm_network_wrapper.py

@@ -40,8 +40,8 @@ def test_initialize_without_graph_crs(self):
         assert isinstance(_wrapper, NetworkWrapperProtocol)
         assert _wrapper.graph_crs.to_epsg() == 4326
 
-    @pytest.fixture
-    def _network_wrapper_without_polygon(self) -> OsmNetworkWrapper:
+    @staticmethod
+    def _get_dummy_network_config_data() -> NetworkConfigData:
         _network_section = NetworkSection(
             network_type=NetworkTypeEnum.DRIVE,
             road_types=[RoadTypeEnum.ROAD],
@@ -50,9 +50,11 @@ def _network_wrapper_without_polygon(self) -> OsmNetworkWrapper:
         _output_dir = test_results.joinpath("test_osm_network_wrapper")
         if not _output_dir.exists():
             _output_dir.mkdir(parents=True)
-        yield OsmNetworkWrapper(
-            NetworkConfigData(network=_network_section, output_path=_output_dir)
-        )
+        return NetworkConfigData(network=_network_section, output_path=_output_dir)
+
+    @pytest.fixture
+    def _network_wrapper_without_polygon(self) -> OsmNetworkWrapper:
+        yield OsmNetworkWrapper(self._get_dummy_network_config_data())
 
     def test_download_clean_graph_from_osm_with_invalid_polygon_arg(
         self, _network_wrapper_without_polygon: OsmNetworkWrapper
@@ -82,17 +84,26 @@ def test_download_clean_graph_from_osm_with_invalid_polygon_arg_geometry(
             == "Geometry must be a shapely Polygon or MultiPolygon. If you requested graph from place name, make sure your query resolves to a Polygon or MultiPolygon, and not some other geometry, like a Point. See OSMnx documentation for details."
         )
 
+    @pytest.mark.parametrize(
+        "road_types",
+        [pytest.param(None, id="With None"), pytest.param([], id="With empty list")],
+    )
     def test_download_clean_graph_from_osm_with_invalid_network_type_arg(
-        self, _network_wrapper_without_polygon: OsmNetworkWrapper
+        self,
+        road_types: list | None,
+        _network_wrapper_without_polygon: OsmNetworkWrapper,
     ):
-        _network_type = "drv"
+        _network_type = NetworkTypeEnum.DRIVE
         _polygon = Polygon([(0.0, 0.0), (0.0, 1.0), (1.0, 1.0), (1.0, 0.0), (0.0, 0.0)])
         with pytest.raises(ValueError) as exc_err:
             _network_wrapper_without_polygon._download_clean_graph_from_osm(
-                _polygon, [""], _network_type
+                _polygon, road_types, _network_type
             )
 
-            assert str(exc_err.value) == f'Unrecognized network_type "{_network_type}"'
+            assert (
+                str(exc_err.value)
+                == f'Unrecognized network_type "{_network_type.config_value}"'
+            )
 
     @pytest.fixture
     def _valid_network_polygon_fixture(self) -> BaseGeometry:
@@ -108,13 +119,13 @@ def test_download_clean_graph_from_osm_output(
         _valid_network_polygon_fixture: BaseGeometry,
     ):
         # 1. Define test data.
-        _link_type = ""
+        _link_type = []
         _network_type = NetworkTypeEnum.DRIVE
 
         # 2. Run test.
         graph_complex = _network_wrapper_without_polygon._download_clean_graph_from_osm(
             polygon=_valid_network_polygon_fixture,
-            network_type=_network_type.config_value,
+            network_type=_network_type,
             road_types=_link_type,
         )
 
@@ -130,26 +141,21 @@ def test_download_clean_graph_from_osm_output(
             map(lambda x: x["osmid"], graph_complex.edges.values())
         )
 
-    def test_get_clean_graph_from_osm_with_invalid_polygon_path_filename(
-        self, _network_wrapper_without_polygon: OsmNetworkWrapper
+    @pytest.mark.parametrize(
+        "polygon_path",
+        [
+            pytest.param(Path("Not_a_valid_path"), id="Not a valid path"),
+            pytest.param(None, id="Path is None"),
+        ],
+    )
+    def test_get_clean_graph_from_osm_with_invalid_polygon_path(
+        self, _network_wrapper_without_polygon: OsmNetworkWrapper, polygon_path: Path
     ):
-        _polygon_path = Path("Not_a_valid_path")
-        _network_wrapper_without_polygon.polygon_path = _polygon_path
-        with pytest.raises(FileNotFoundError) as exc_err:
-            _network_wrapper_without_polygon.get_clean_graph_from_osm()
-
-        assert str(exc_err.value) == "No polygon_file file found at {}.".format(
-            _polygon_path
+        _result = _network_wrapper_without_polygon._get_clean_graph_from_osm(
+            polygon_path
         )
 
-    def test_get_clean_graph_from_osm_with_no_polygon_path(
-        self, _network_wrapper_without_polygon: OsmNetworkWrapper
-    ):
-        _network_wrapper_without_polygon.polygon_path = None
-        with pytest.raises(ValueError) as exc_err:
-            _network_wrapper_without_polygon.get_clean_graph_from_osm()
-
-        assert str(exc_err.value) == "No valid value provided for polygon file."
+        assert _result is None
 
     @pytest.fixture
     def _valid_graph_fixture(self) -> MultiDiGraph:
@@ -277,3 +283,21 @@ def test_snap_nodes_to_nodes(self, _valid_graph_fixture: MultiDiGraph):
 
         # 3. Verify expectations.
         assert isinstance(_result_graph, MultiDiGraph)
+
+    @slow_test
+    def test_given_valid_base_geometry_with_polygon(
+        self, _valid_network_polygon_fixture: BaseGeometry
+    ):
+        # 1. Define test data.
+        _network_config_data = self._get_dummy_network_config_data()
+        _network_config_data.network.network_type = NetworkTypeEnum.DRIVE
+        _network_config_data.network.road_types = []
+
+        # 2. Run test.
+        _wrapper = OsmNetworkWrapper.with_polygon(
+            _network_config_data, _valid_network_polygon_fixture
+        )
+
+        # 3. Verify expectations.
+        assert isinstance(_wrapper, OsmNetworkWrapper)
+        assert isinstance(_wrapper.polygon_graph, MultiDiGraph)