update code

Author: debpal

GeoAnalyze/stream.py

@@ -55,7 +55,6 @@ def is_flw_path_us_to_ds(
 
         return output
 
-    # pytest pending
     def flw_path_reverse(
         self,
         input_file: str,
@@ -270,7 +269,6 @@ def point_junctions(
 
         return output_gdf
 
-    # pytest pending
     def point_segment_subbasin_drainage(
         self,
         input_file: str,
@@ -391,8 +389,7 @@ def point_main_outlets(
 
         return outlet_gdf
 
-    # pytest pending
-    def create_box_touching_selected_segment(
+    def box_touch_selected_segment(
         self,
         input_file: str,
         column_name: str,
@@ -402,8 +399,8 @@ def create_box_touching_selected_segment(
     ) -> geopandas.GeoDataFrame:
 
         '''
-        Creates a square box polygon that touches a specified stream segment
-        in the input flow path at a randomly chosen point along the segment.
+        Creates a square box polygon that touches a specified segment
+        in the stream path at a randomly chosen point along the segment.
 
         Parameters
         ----------
@@ -429,17 +426,19 @@ def create_box_touching_selected_segment(
             touches the specified stream segment at a random point.
         '''
 
+        # check validity of output file path
+        check_file = Core().is_valid_ogr_driver(output_file)
+        if check_file is True:
+            pass
+        else:
+            raise Exception('Could not retrieve driver from the file path.')
+
         # input line segment
         gdf = geopandas.read_file(input_file)
         line = gdf[gdf[column_name].isin([column_value])].geometry.iloc[0]
 
         # line coords
-        if isinstance(line, shapely.LineString):
-            line_coords = line.coords[:]
-        else:
-            line_coords = []
-            for line_split in line.geoms:
-                line_coords.extend(line_split.coords[:])
+        line_coords = line.coords[:] if isinstance(line, shapely.LineString) else [c for ls in line.geoms for c in ls.coords[:]]
 
         while True:
             # choose points
@@ -464,8 +463,6 @@ def create_box_touching_selected_segment(
             check_touch = line.touches(rotate_box) and not line.crosses(rotate_box)
             if check_touch is True:
                 break
-            else:
-                pass
 
         # saving box geodataframe
         box_gdf = geopandas.GeoDataFrame(
@@ -476,20 +473,19 @@ def create_box_touching_selected_segment(
 
         return box_gdf
 
-    # pytest pending
-    def create_box_crosses_segment_at_endpoint(
+    def box_touch_selected_segment_at_endpoint(
         self,
         input_file: str,
         column_name: str,
         column_value: typing.Any,
         box_length: float,
         output_file: str,
-        downstream_point: bool = True
+        upstream_point: bool = True
     ) -> geopandas.GeoDataFrame:
 
         '''
-        Creates a square box polygon that crosses a specified stream segment
-        in the input flow path and passes through an endpoint of the segment.
+        Creates a square box polygon that touches an endpoint
+        of a specified segment in the input stream path.
 
         Parameters
         ----------
@@ -508,43 +504,50 @@ def create_box_crosses_segment_at_endpoint(
         output_file : str
             Path to save the output box polygon shapefile.
 
-        downstream_point : bool, optional
-            If True, the box is positioned to pass through the downstream endpoint
-            of the segment; if False, it passes through the upstream endpoint. Default is True.
+        upstream_point : bool, optional
+            If True, the box is positioned to pass through the upstream endpoint
+            of the segment; if False, it passes through the downstream endpoint. Default is True.
 
         Returns
         -------
         GeoDataFrame
-            A GeoDataFrame containing the box polygon, which crosses the
-            specified stream segment and passes through an endpoint of the segment.
+            A GeoDataFrame containing the box polygon, which touches an endpoint of
+            the specified segment in the input stream path.
         '''
 
+        # check validity of output file path
+        check_file = Core().is_valid_ogr_driver(output_file)
+        if check_file is True:
+            pass
+        else:
+            raise Exception('Could not retrieve driver from the file path.')
+
         # input line segement
         gdf = geopandas.read_file(input_file)
         line = gdf[gdf[column_name].isin([column_value])].geometry.iloc[0]
 
         # get point
-        point_coords = line.coords[-1] if downstream_point is True else line.coords[0]
+        point_coords = line.coords[0] if upstream_point is True else line.coords[-1]
         point = shapely.Point(*point_coords)
 
         # create box
         box = shapely.box(
             xmin=point.x,
             ymin=point.y,
             xmax=point.x + box_length,
-            ymax=point.y + box_length,
+            ymax=point.y + box_length
         )
 
-        # check whether the box crosses the line; otherwise rotate
+        # check whether the box touches the line; otherwise rotate
         while True:
-            if line.crosses(box):
+            box = shapely.affinity.rotate(
+                geom=box,
+                angle=random.randint(0, 360),
+                origin=point
+            )
+            check_touch = line.touches(box) and not line.crosses(box)
+            if check_touch:
                 break
-            else:
-                box = shapely.affinity.rotate(
-                    geom=box,
-                    angle=random.randint(0, 360),
-                    origin=point
-                )
 
         # saving box geodataframe
         box_gdf = geopandas.GeoDataFrame(
@@ -555,20 +558,19 @@ def create_box_crosses_segment_at_endpoint(
 
         return box_gdf
 
-    # pytest pending
-    def create_box_touch_segment_at_endpoint(
+    def box_cross_selected_segment_at_endpoint(
         self,
         input_file: str,
         column_name: str,
         column_value: typing.Any,
         box_length: float,
         output_file: str,
-        upstream_point: bool = True
+        downstream_point: bool = True
     ) -> geopandas.GeoDataFrame:
 
         '''
-        Creates a square box polygon that touches an endpoint
-        of a specified segment in the input stream path.
+        Creates a square box polygon that crosses a specified segment
+        in the stream path and passes through an endpoint of the segment.
 
         Parameters
         ----------
@@ -587,44 +589,49 @@ def create_box_touch_segment_at_endpoint(
         output_file : str
             Path to save the output box polygon shapefile.
 
-        upstream_point : bool, optional
-            If True, the box is positioned to pass through the upstream endpoint
-            of the segment; if False, it passes through the downstream endpoint. Default is True.
+        downstream_point : bool, optional
+            If True, the box is positioned to pass through the downstream endpoint
+            of the segment; if False, it passes through the upstream endpoint. Default is True.
 
         Returns
         -------
         GeoDataFrame
-            A GeoDataFrame containing the box polygon, which touches an endpoint of
-            the specified segment in the input stream path.
+            A GeoDataFrame containing the box polygon, which crosses the
+            specified stream segment and passes through an endpoint of the segment.
         '''
 
+        # check validity of output file path
+        check_file = Core().is_valid_ogr_driver(output_file)
+        if check_file is True:
+            pass
+        else:
+            raise Exception('Could not retrieve driver from the file path.')
+
         # input line segement
         gdf = geopandas.read_file(input_file)
         line = gdf[gdf[column_name].isin([column_value])].geometry.iloc[0]
 
         # get point
-        point_coords = line.coords[0] if upstream_point is True else line.coords[-1]
+        point_coords = line.coords[-1] if downstream_point is True else line.coords[0]
         point = shapely.Point(*point_coords)
 
         # create box
         box = shapely.box(
             xmin=point.x,
             ymin=point.y,
             xmax=point.x + box_length,
-            ymax=point.y + box_length,
+            ymax=point.y + box_length
         )
 
-        # check whether the box touches the line; otherwise rotate
+        # check whether the box crosses the line; otherwise rotate
         while True:
-            check_touch = line.touches(box) and not line.crosses(box)
-            if check_touch:
+            box = shapely.affinity.rotate(
+                geom=box,
+                angle=random.randint(0, 360),
+                origin=point
+            )
+            if line.crosses(box):
                 break
-            else:
-                box = shapely.affinity.rotate(
-                    geom=box,
-                    angle=random.randint(0, 360),
-                    origin=point
-                )
 
         # saving box geodataframe
         box_gdf = geopandas.GeoDataFrame(

tests/test_stream.py

@@ -91,6 +91,38 @@ def test_functions(
             output_file=os.path.join(tmp_dir, 'main_outlet_points.shp')
         )
         assert stream_gdf['geometry'].iloc[-1].coords[-1] == outlet_gdf['geometry'].iloc[-1].coords[0]
+        # pass test of box touching the selected segment in a stream path
+        selected_line = stream_gdf[stream_gdf['flw_id'] == 3]['geometry'].iloc[0]
+        box_gdf = stream.box_touch_selected_segment(
+            input_file=stream_file,
+            column_name='flw_id',
+            column_value=3,
+            box_length=500,
+            output_file=os.path.join(tmp_dir, 'box.shp')
+        )
+        polygon = box_gdf.geometry.iloc[0]
+        assert selected_line.touches(polygon)
+        # pass test of box touching the selected segment at endpoint in a stream path
+        box_gdf = stream.box_touch_selected_segment_at_endpoint(
+            input_file=stream_file,
+            column_name='flw_id',
+            column_value=3,
+            box_length=500,
+            output_file=os.path.join(tmp_dir, 'box.shp')
+        )
+        polygon = box_gdf.geometry.iloc[0]
+        assert selected_line.touches(polygon)
+        # pass test of box crossing the selected segment at endpoint in a stream path
+        box_gdf = stream.box_cross_selected_segment_at_endpoint(
+            input_file=stream_file,
+            column_name='flw_id',
+            column_value=3,
+            box_length=500,
+            output_file=os.path.join(tmp_dir, 'box.shp')
+        )
+        polygon = box_gdf.geometry.iloc[0]
+        intersection = selected_line.intersection(polygon)
+        assert isinstance(intersection, shapely.MultiLineString) or len(intersection.coords[:]) > 1
 
 
 def test_error_linestring_geometry(
@@ -190,3 +222,33 @@ def test_error_shapefile_driver(
             output_file='main_outlet_points.sh'
         )
     assert exc_info.value.args[0] == message['error_driver']
+    # box touching the selected segment in a stream path
+    with pytest.raises(Exception) as exc_info:
+        stream.box_touch_selected_segment(
+            input_file='stream.shp',
+            column_name='flw_id',
+            column_value=3,
+            box_length=500,
+            output_file='box.sh'
+        )
+    assert exc_info.value.args[0] == message['error_driver']
+    # box touching the selected segment at endpoint in a stream path
+    with pytest.raises(Exception) as exc_info:
+        stream.box_touch_selected_segment_at_endpoint(
+            input_file='stream.shp',
+            column_name='flw_id',
+            column_value=3,
+            box_length=500,
+            output_file='box.sh'
+        )
+    assert exc_info.value.args[0] == message['error_driver']
+    # box crossing the selected segment at endpoint in a stream path
+    with pytest.raises(Exception) as exc_info:
+        stream.box_cross_selected_segment_at_endpoint(
+            input_file='stream.shp',
+            column_name='flw_id',
+            column_value=3,
+            box_length=500,
+            output_file='box.sh'
+        )
+    assert exc_info.value.args[0] == message['error_driver']