Few more modifications

Author: sujata-m

CHANGELOG.md

@@ -2,6 +2,8 @@
 - Fixed [Task-1347](https://dev.azure.com/TDEI-UW/TDEI/_workitems/edit/1347/).
 - Fixed package to removing the additional keys from the geojson files. 
 - Introduced garbage collection to free up memory.
+- Added ability to skip the tags which are already present in the edges file.
+- Added ability to process the incline tags in batch processing.
 
 
 ### 0.0.1

README.md

@@ -47,6 +47,12 @@ osw_incline = OSWIncline(
 # Perform the incline calculation, it will add the incline to the  original edges file 
 result = osw_incline.calculate()
 
+# To skip the incline tags which are already present in the edges file
+result = osw_incline.calculate(skip_existing_tags=True)
+
+# To update the incline tags in batch processing (It might be faster than the normal calculation but increases the memory usage)
+result = osw_incline.calculate(batch_processing=True)
+
 if result:
     print("Incline calculation completed successfully.")
 ```

src/osw_incline/__init__.py

@@ -17,7 +17,7 @@ def __init__(self, dem_files: List[str], nodes_file: str, edges_file: str, debug
         if self.debug:
             Logger.debug('Debug mode is enabled')
 
-    def calculate(self):
+    def calculate(self, skip_existing_tags=False, batch_processing=False):
         try:
             if self.debug:
                 Logger.debug('Starting calculation process')
@@ -33,10 +33,13 @@ def calculate(self):
             dem_processor = DEMProcessor(osm_graph=osm_graph, dem_files=self.dem_files, debug=self.debug)
             dem_processor.process(
                 nodes_path=graph_nodes_path,
-                edges_path=graph_edges_path
+                edges_path=graph_edges_path,
+                skip_existing_tags=skip_existing_tags,
+                batch_processing=batch_processing
             )
 
             # Delete osm_graph and dem_processor to force garbage collection
+            osm_graph.clean()
             del osm_graph, dem_processor
             gc.collect()
 

src/osw_incline/dem_processor.py

@@ -22,7 +22,7 @@ def __init__(self, osm_graph: OSMGraph, dem_files: List[str], debug=False):
         self.OG = osm_graph
         self.debug = debug
 
-    def process(self, nodes_path, edges_path):
+    def process(self, nodes_path, edges_path, skip_existing_tags=False, batch_processing=False):
         gc.disable()
         for dem_file in self.dem_files:
             dem_file_path = Path(dem_file)
@@ -31,8 +31,41 @@ def process(self, nodes_path, edges_path):
 
             try:
                 with rasterio.open(dem_file_path) as dem:
-                    edges = list(self.OG.G.edges(data=True))  # Get all edges, even if fewer than batch_size
-                    self._process_in_batches(edges, dem, batch_size=1000)
+                    """
+                    Option 1:
+                        Pros:
+                            Batching: This approach processes edges in batches of 1000, which can be faster for large graphs.
+                            Parallelization: The second approach can be parallelized by using a ThreadPoolExecutor or similar.
+                        Cons:
+                            Memory usage: The second approach stores all edges in a list, which could be memory-intensive for large graphs.
+                            Intermediate list storage: The second approach stores the entire edge set as a list in memory, which is not memory-efficient.
+                    """
+                    if batch_processing:
+                        edges = list(self.OG.G.edges(data=True))  # Get all edges, even if fewer than batch_size
+                        self._process_in_batches(edges, dem, batch_size=1000, skip_existing_tags=skip_existing_tags)
+                    else:
+                        """
+                        Option 2:
+                            Pros:
+                                Simple iteration: The first approach iterates over the edges one by one, making the memory footprint relatively small, especially if you have a large number of edges.
+                                No intermediate list storage: It does not store the entire edge set as a list in memory, which is better for memory efficiency.
+                            Cons:
+                                Single-threaded: The entire edge processing happens sequentially, which can be slower for very large graphs, as there's no batching or parallelization.
+                                No batching: It processes all edges at once in a loop, which could cause memory spikes during large computations if infer_incline holds intermediate states or large datasets.
+                        """
+                        for u, v, d in self.OG.G.edges(data=True):
+                            if 'geometry' in d:
+                                if skip_existing_tags:
+                                    if 'incline' in d and d['incline'] is not None:
+                                        # If incline already exists, skip
+                                        continue
+                                incline = self.infer_incline(linestring=d['geometry'], dem=dem, precision=3)
+                                if incline is not None:
+                                    # Add incline to the edge properties
+                                    d['incline'] = incline
+                            else:
+                                if self.debug:
+                                    Logger.info(f'No geometry found for edge {u}-{v}')
 
                 self.OG.to_geojson(nodes_path, edges_path)
             except rasterio.errors.RasterioIOError:
@@ -48,12 +81,16 @@ def process(self, nodes_path, edges_path):
 
         gc.disable()
 
-    def _process_in_batches(self, edges, dem, batch_size=1000):
+    def _process_in_batches(self, edges, dem, batch_size=1000, skip_existing_tags=False):
         # Process edges in batches
         for i in range(0, len(edges), batch_size):
             batch = edges[i:i + batch_size]
             for u, v, d in batch:
                 if 'geometry' in d:
+                    if skip_existing_tags:
+                        if 'incline' in d and d['incline'] is not None:
+                            # If incline already exists, skip
+                            continue
                     incline = self.infer_incline(linestring=d['geometry'], dem=dem, precision=3)
                     if incline is not None:
                         d['incline'] = incline

src/osw_incline/osm_graph.py

@@ -151,4 +151,8 @@ def to_geojson(self, *args):
 
             # Delete point_features and force garbage collection
             del point_features, points_fc
-            gc.collect()
+            gc.collect()
+
+    def clean(self):
+        del self.G
+        gc.collect()

tests/test_dem_processor.py

@@ -39,6 +39,28 @@ def test_process_success(self, mock_rasterio_open):
 
         self.osm_graph.to_geojson.assert_called_once_with('nodes.json', 'edges.json')
 
+    @patch('src.osw_incline.dem_processor.rasterio.open')
+    def test_process_success_with_batching(self, mock_rasterio_open):
+        mock_dem = MagicMock()
+        mock_rasterio_open.return_value.__enter__.return_value = mock_dem
+        self.osm_graph.G.edges.return_value = [('u', 'v', {'geometry': LineString([(0, 0), (1, 1)])})]
+
+        with patch.object(self.processor, 'infer_incline', return_value=0.1):
+            self.processor.process('nodes.json', 'edges.json', batch_processing=True)
+
+        self.osm_graph.to_geojson.assert_called_once_with('nodes.json', 'edges.json')
+
+    @patch('src.osw_incline.dem_processor.rasterio.open')
+    def test_process_success_with_skip(self, mock_rasterio_open):
+        mock_dem = MagicMock()
+        mock_rasterio_open.return_value.__enter__.return_value = mock_dem
+        self.osm_graph.G.edges.return_value = [('u', 'v', {'geometry': LineString([(0, 0), (1, 1)])})]
+
+        with patch.object(self.processor, 'infer_incline', return_value=0.1):
+            self.processor.process('nodes.json', 'edges.json', skip_existing_tags=True)
+
+        self.osm_graph.to_geojson.assert_called_once_with('nodes.json', 'edges.json')
+
     # Test processing when RasterioIOError is raised
     @patch('src.osw_incline.dem_processor.rasterio.open')
     def test_process_rasterio_io_error(self, mock_rasterio_open):

tests/test_osw_incline.py

@@ -44,11 +44,78 @@ def test_calculate_success(self, mock_logger_info, mock_time, mock_dem_processor
         )
         mock_dem_processor.assert_called_once_with(
             nodes_path=Path(self.nodes_file),
-            edges_path=Path(self.edges_file)
+            edges_path=Path(self.edges_file),
+            skip_existing_tags=False,
+            batch_processing=False
+        )
+
+    @patch.object(OSMGraph, 'from_geojson', return_value=MagicMock())
+    @patch('src.osw_incline.dem_processor.DEMProcessor.process', return_value=None)
+    @patch('time.time', side_effect=[1, 5])  # Simulate time taken for the calculation
+    @patch.object(Logger, 'info')  # Mock the Logger to capture log calls
+    def test_calculate_success_with_skip_existing_tags(self, mock_logger_info, mock_time, mock_dem_processor, mock_osm_graph):
+        result = self.osw_incline.calculate(skip_existing_tags=True)
+
+        # Check if the process was successful
+        self.assertTrue(result)
+
+        # Ensure the OSMGraph and DEMProcessor were used correctly
+        mock_osm_graph.assert_called_once_with(
+            nodes_path=Path(self.nodes_file),
+            edges_path=Path(self.edges_file),
+        )
+        mock_dem_processor.assert_called_once_with(
+            nodes_path=Path(self.nodes_file),
+            edges_path=Path(self.edges_file),
+            skip_existing_tags=True,
+            batch_processing=False
+        )
+
+    @patch.object(OSMGraph, 'from_geojson', return_value=MagicMock())
+    @patch('src.osw_incline.dem_processor.DEMProcessor.process', return_value=None)
+    @patch('time.time', side_effect=[1, 5])  # Simulate time taken for the calculation
+    @patch.object(Logger, 'info')  # Mock the Logger to capture log calls
+    def test_calculate_success_with_batch_processing(self, mock_logger_info, mock_time, mock_dem_processor,
+                                                       mock_osm_graph):
+        result = self.osw_incline.calculate(batch_processing=True)
+
+        # Check if the process was successful
+        self.assertTrue(result)
+
+        # Ensure the OSMGraph and DEMProcessor were used correctly
+        mock_osm_graph.assert_called_once_with(
+            nodes_path=Path(self.nodes_file),
+            edges_path=Path(self.edges_file),
         )
+        mock_dem_processor.assert_called_once_with(
+            nodes_path=Path(self.nodes_file),
+            edges_path=Path(self.edges_file),
+            skip_existing_tags=False,
+            batch_processing=True
+        )
+
+    @patch.object(OSMGraph, 'from_geojson', return_value=MagicMock())
+    @patch('src.osw_incline.dem_processor.DEMProcessor.process', return_value=None)
+    @patch('time.time', side_effect=[1, 5])  # Simulate time taken for the calculation
+    @patch.object(Logger, 'info')  # Mock the Logger to capture log calls
+    def test_calculate_success_with_batching_and_skip_existing_tags(self, mock_logger_info, mock_time, mock_dem_processor,
+                                                     mock_osm_graph):
+        result = self.osw_incline.calculate(skip_existing_tags=True, batch_processing=True)
+
+        # Check if the process was successful
+        self.assertTrue(result)
 
-        # Check if the time taken was logged
-        mock_logger_info.assert_called_once_with('Entire processing took: 4 seconds')
+        # Ensure the OSMGraph and DEMProcessor were used correctly
+        mock_osm_graph.assert_called_once_with(
+            nodes_path=Path(self.nodes_file),
+            edges_path=Path(self.edges_file),
+        )
+        mock_dem_processor.assert_called_once_with(
+            nodes_path=Path(self.nodes_file),
+            edges_path=Path(self.edges_file),
+            skip_existing_tags=True,
+            batch_processing=True
+        )
 
     # Test when OSMGraph.from_geojson raises an exception
     @patch.object(OSMGraph, 'from_geojson', side_effect=Exception("OSMGraph Error"))