Dev to Main

.github/workflows/unit_tests.yml

@@ -5,7 +5,7 @@ on:
     branches-ignore:
       - '**'
   pull_request:
-    branches: [main, dev, stage]
+    branches: [dev]
 
 jobs:
   UnitTest:
@@ -27,7 +27,7 @@ jobs:
 
     - name: Run tests with coverage
       run: |
-        coverage run --source=src -m unittest discover -s tests/
+        coverage run --source=src/osw_incline -m unittest discover -s tests/
         coverage xml  
 
     - name: Check coverage

CHANGELOG.md

@@ -1,12 +1,19 @@
+### 0.0.3
+
+- Fixed [Task-1369](https://dev.azure.com/TDEI-UW/TDEI/_workitems/edit/1369/).
+- Updated the code to handle incline less than -1 ot greater than 1.
+- Updated unit test cases
+
 ### 0.0.2
+
 - Fixed [Task-1347](https://dev.azure.com/TDEI-UW/TDEI/_workitems/edit/1347/).
-- Fixed package to removing the additional keys from the geojson files. 
+- 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
+
 - Introduces osw_inclination package which calculates the inclination of the sidewalk based on the DEM data.
 - Added example.py file which demonstrates how to use the package.
 - Added unit test cases.

README.md

@@ -99,7 +99,7 @@ OSW-Incline includes a suite of unit tests to ensure correct functionality. You
 python -m unittest discover -v tests
 
 # To run the unit test cases with coverage
-python -m coverage run --source=src -m unittest discover -v tests
+python -m coverage run --source=src/osw_incline -m unittest discover -v tests
 
 # To generate the coverage report
 python -m coverage report

requirements.txt

@@ -5,4 +5,5 @@ rasterio
 numpy
 scipy
 jinja2
-coverage
+coverage
+requests

src/example.py

@@ -1,25 +1,23 @@
 import os
-
-from jinja2.ext import debug
-
 from osw_incline import OSWIncline
-
+from utils import download_dems, unzip_dataset, remove_unzip_dataset
 
 PARENT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
-DOWNLOADS_DIR = os.path.join(PARENT_DIR, 'downloads')
+DEM_DIR = os.path.join(PARENT_DIR, 'downloads/dems')
+ASSETS_DIR = os.path.join(PARENT_DIR, 'tests/assets')
 
 
 def test_incline():
-    dem_files = [f'{DOWNLOADS_DIR}/dems/n48w123.tif']
-    nodes_file = f'{DOWNLOADS_DIR}/geojson_renton_hth/renton_hth.nodes.geojson'
-    edges_file = f'{DOWNLOADS_DIR}/geojson_renton_hth/renton_hth.edges.geojson'
+    dem_files = [f'{DEM_DIR}/n48w123.tif']
+    nodes_file = f'{ASSETS_DIR}/medium/wa.seattle.graph.nodes.geojson'
+    edges_file = f'{ASSETS_DIR}/medium/wa.seattle.graph.edges.geojson'
     incline = OSWIncline(dem_files=dem_files, nodes_file=nodes_file, edges_file=edges_file, debug=True)
     result = incline.calculate()
     print(result)
 
 
-
-
 if __name__ == '__main__':
+    download_dems()
+    unzip_dataset()
     test_incline()
-
+    remove_unzip_dataset()

src/osw_incline/dem_processor.py

@@ -42,7 +42,7 @@ def process(self, nodes_path, edges_path, skip_existing_tags=False, batch_proces
                     """
                     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)
+                        self._process_in_batches(edges, dem, batch_size=10000, skip_existing_tags=skip_existing_tags)
                     else:
                         """
                         Option 2:
@@ -57,10 +57,12 @@ def process(self, nodes_path, edges_path, skip_existing_tags=False, batch_proces
                             if 'geometry' in d:
                                 if skip_existing_tags:
                                     if 'incline' in d and d['incline'] is not None:
+                                        if -1 <= d['incline'] <= 1:
+                                            del d['incline']
                                         # If incline already exists, skip
                                         continue
                                 incline = self.infer_incline(linestring=d['geometry'], dem=dem, precision=3)
-                                if incline is not None:
+                                if incline is not None and -1 <= incline <= 1:
                                     # Add incline to the edge properties
                                     d['incline'] = incline
                             else:
@@ -81,18 +83,20 @@ def process(self, nodes_path, edges_path, skip_existing_tags=False, batch_proces
 
         gc.disable()
 
-    def _process_in_batches(self, edges, dem, batch_size=1000, skip_existing_tags=False):
+    def _process_in_batches(self, edges, dem, batch_size=10000, 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 -1 <= d['incline'] <= 1:
+                                del d['incline']
                             # If incline already exists, skip
                             continue
                     incline = self.infer_incline(linestring=d['geometry'], dem=dem, precision=3)
-                    if incline is not None:
+                    if incline is not None and -1 <= incline <= 1:
                         d['incline'] = incline
             # Trigger garbage collection after each batch
             gc.collect()

src/osw_incline/version.py

@@ -1 +1 @@
-__version__ = '0.0.2'
+__version__ = '0.0.3'

src/utils/__init__.py

@@ -0,0 +1,40 @@
+import shutil
+import zipfile
+import requests
+from pathlib import Path
+
+DOWNLOAD_DIR = Path(f'{Path.cwd()}/downloads')
+DOWNLOAD_DIR.mkdir(parents=True, exist_ok=True)
+DEM_DIR = Path(DOWNLOAD_DIR, 'dems')
+DEM_DIR.mkdir(parents=True, exist_ok=True)
+
+ASSETS_DIR = Path(f'{Path.cwd()}/tests/assets')
+
+
+def download_dems():
+    dem_file = 'n48w123.tif'
+    file_path = Path(f'{DEM_DIR}/{dem_file}')
+
+    # Download DEM file if not present
+    if not file_path.is_file():
+        URL = 'https://prd-tnm.s3.amazonaws.com/StagedProducts/Elevation/13/TIFF/current/n48w123/USGS_13_n48w123.tif'
+        with requests.get(URL, stream=True) as r:
+            r.raise_for_status()
+            with open(file_path, 'wb') as f:
+                for chunk in r.iter_content(chunk_size=8192):
+                    f.write(chunk)
+
+
+def unzip_dataset():
+    zip_path = Path(f'{ASSETS_DIR}/medium.zip')
+    extract_to = Path(f'{ASSETS_DIR}/medium')
+
+    # Unzip medium.zip to the specified extraction path
+    with zipfile.ZipFile(zip_path, 'r') as zip_ref:
+        zip_ref.extractall(extract_to)
+
+
+def remove_unzip_dataset():
+    extract_to = Path(f'{ASSETS_DIR}/medium')
+    if extract_to.exists():
+        shutil.rmtree(extract_to, ignore_errors=True)

tests/test_osw_incline.py

@@ -1,9 +1,14 @@
+import json
 import unittest
 from pathlib import Path
 from src.osw_incline import OSWIncline
 from src.osw_incline.logger import Logger
 from unittest.mock import patch, MagicMock
 from src.osw_incline.osm_graph import OSMGraph
+from src.utils import download_dems, unzip_dataset, remove_unzip_dataset
+
+ASSETS_DIR = f'{Path.cwd()}/tests/assets'
+DEM_DIR = f'{Path.cwd()}/downloads/dems'
 
 
 class TestOSWIncline(unittest.TestCase):
@@ -53,7 +58,8 @@ def test_calculate_success(self, mock_logger_info, mock_time, mock_dem_processor
     @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):
+    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
@@ -76,7 +82,7 @@ def test_calculate_success_with_skip_existing_tags(self, mock_logger_info, mock_
     @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):
+                                                     mock_osm_graph):
         result = self.osw_incline.calculate(batch_processing=True)
 
         # Check if the process was successful
@@ -98,8 +104,9 @@ def test_calculate_success_with_batch_processing(self, mock_logger_info, mock_ti
     @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):
+    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
@@ -169,5 +176,49 @@ def test_debug_logging_on_initialization(self, mock_logger_debug):
         mock_logger_debug.assert_called_once_with('Debug mode is enabled')
 
 
+class TestOSWInclineIntegration(unittest.TestCase):
+    def setUp(self):
+        self.dem_files = [f'{DEM_DIR}/n48w123.tif']
+        # Download DEM file if not present
+        download_dems()
+        # unzip medium.zip to the specified extraction path
+        unzip_dataset()
+        self.extract_to = Path(f'{ASSETS_DIR}/medium')
+        self.nodes_file = f'{self.extract_to}/wa.seattle.graph.nodes.geojson'
+        self.edges_file = f'{self.extract_to}/wa.seattle.graph.edges.geojson'
+
+    def tearDown(self):
+        remove_unzip_dataset()
+
+    def test_entire_process(self):
+        incline = OSWIncline(
+            dem_files=self.dem_files,
+            nodes_file=str(self.nodes_file),
+            edges_file=str(self.edges_file),
+            debug=True
+        )
+        result = incline.calculate()
+        self.assertTrue(result)
+
+    def test_incline_tag_added(self):
+        incline = OSWIncline(
+            dem_files=self.dem_files,
+            nodes_file=self.nodes_file,
+            edges_file=self.edges_file,
+            debug=True
+        )
+        result = incline.calculate()
+        self.assertTrue(result)
+
+        with open(self.edges_file, 'r') as f:
+            edges_data = json.load(f)
+
+        for feature in edges_data['features']:
+            if 'incline' in feature['properties']:
+                incline_value = feature['properties']['incline']
+                self.assertIsInstance(incline_value, (int, float), 'Incline should be an integer or float.')
+                self.assertTrue(-1 <= incline_value <= 1, 'Incline should be between -1 and 1.')
+
+
 if __name__ == '__main__':
     unittest.main()