reference lines, points

reference points geom

reflines, refpoints output

tests for reflines, refpoints

docs/source/RasGeomHdf.rst

@@ -12,6 +12,8 @@ RasGeomHdf
              bc_lines,
              breaklines,
              refinement_regions,
+             reference_lines,
+             reference_points,
              structures,
              get_geom_attrs,
              get_geom_structures_attrs,

docs/source/RasPlanHdf.rst

@@ -13,6 +13,8 @@ RasPlanHdf
              mesh_max_ws_err,
              mesh_max_iter,
              mesh_last_iter,
+             reference_lines,
+             reference_points,
              get_plan_info_attrs,
              get_plan_param_attrs,
              get_meteorology_precip_attrs,

src/cli.py

@@ -23,6 +23,8 @@
     "refinement_regions",
     "bc_lines",
     "breaklines",
+    "reference_lines",
+    "reference_points",
     "structures",
 ]
 

src/rashdf/geom.py

@@ -13,6 +13,7 @@
 from geopandas import GeoDataFrame
 from pyproj import CRS
 from shapely import (
+    Geometry,
     Polygon,
     Point,
     LineString,
@@ -21,7 +22,7 @@
     polygonize_full,
 )
 
-from typing import Dict, List, Optional
+from typing import Dict, List, Optional, Union
 
 
 class RasGeomHdf(RasHdf):
@@ -30,6 +31,10 @@ class RasGeomHdf(RasHdf):
     GEOM_PATH = "Geometry"
     GEOM_STRUCTURES_PATH = f"{GEOM_PATH}/Structures"
     FLOW_AREA_2D_PATH = f"{GEOM_PATH}/2D Flow Areas"
+    BC_LINES_PATH = f"{GEOM_PATH}/Boundary Condition Lines"
+    BREAKLINES_PATH = f"{GEOM_PATH}/2D Flow Area Break Lines"
+    REFERENCE_LINES_PATH = f"{GEOM_PATH}/Reference Lines"
+    REFERENCE_POINTS_PATH = f"{GEOM_PATH}/Reference Points"
 
     def __init__(self, name: str, **kwargs):
         """Open a HEC-RAS Geometry HDF file.
@@ -262,6 +267,38 @@ def get_geom_2d_flow_area_attrs(self) -> Dict:
 
         return d2_flow_area_attrs
 
+    def _get_polylines(
+        self,
+        path: str,
+        info_name: str = "Polyline Info",
+        parts_name: str = "Polyline Parts",
+        points_name: str = "Polyline Points",
+    ) -> List[Geometry]:
+        polyline_info_path = f"{path}/{info_name}"
+        polyline_parts_path = f"{path}/{parts_name}"
+        polyline_points_path = f"{path}/{points_name}"
+
+        polyline_info = self[polyline_info_path][()]
+        polyline_parts = self[polyline_parts_path][()]
+        polyline_points = self[polyline_points_path][()]
+
+        geoms = []
+        for pnt_start, pnt_cnt, part_start, part_cnt in polyline_info:
+            points = polyline_points[pnt_start : pnt_start + pnt_cnt]
+            if part_cnt == 1:
+                geoms.append(LineString(points))
+            else:
+                parts = polyline_parts[part_start : part_start + part_cnt]
+                geoms.append(
+                    MultiLineString(
+                        list(
+                            points[part_pnt_start : part_pnt_start + part_pnt_cnt]
+                            for part_pnt_start, part_pnt_cnt in parts
+                        )
+                    )
+                )
+        return geoms
+
     def bc_lines(self) -> GeoDataFrame:
         """Return 2D mesh area boundary condition lines.
 
@@ -270,35 +307,15 @@ def bc_lines(self) -> GeoDataFrame:
         GeoDataFrame
             A GeoDataFrame containing the 2D mesh area boundary condition lines if they exist.
         """
-        if "/Geometry/Boundary Condition Lines" not in self:
+        if self.BC_LINES_PATH not in self:
             return GeoDataFrame()
-        bc_line_data = self["/Geometry/Boundary Condition Lines"]
+        bc_line_data = self[self.BC_LINES_PATH]
         bc_line_ids = range(bc_line_data["Attributes"][()].shape[0])
         v_conv_str = np.vectorize(convert_ras_hdf_string)
         names = v_conv_str(bc_line_data["Attributes"][()]["Name"])
         mesh_names = v_conv_str(bc_line_data["Attributes"][()]["SA-2D"])
         types = v_conv_str(bc_line_data["Attributes"][()]["Type"])
-        geoms = list()
-        for pnt_start, pnt_cnt, part_start, part_cnt in bc_line_data["Polyline Info"][
-            ()
-        ]:
-            points = bc_line_data["Polyline Points"][()][
-                pnt_start : pnt_start + pnt_cnt
-            ]
-            if part_cnt == 1:
-                geoms.append(LineString(points))
-            else:
-                parts = bc_line_data["Polyline Parts"][()][
-                    part_start : part_start + part_cnt
-                ]
-                geoms.append(
-                    MultiLineString(
-                        list(
-                            points[part_pnt_start : part_pnt_start + part_pnt_cnt]
-                            for part_pnt_start, part_pnt_cnt in parts
-                        )
-                    )
-                )
+        geoms = self._get_polylines(self.BC_LINES_PATH)
         return GeoDataFrame(
             {
                 "bc_line_id": bc_line_ids,
@@ -319,34 +336,14 @@ def breaklines(self) -> GeoDataFrame:
         GeoDataFrame
             A GeoDataFrame containing the 2D mesh area breaklines if they exist.
         """
-        if "/Geometry/2D Flow Area Break Lines" not in self:
+        if self.BREAKLINES_PATH not in self:
             return GeoDataFrame()
-        bl_line_data = self["/Geometry/2D Flow Area Break Lines"]
+        bl_line_data = self[self.BREAKLINES_PATH]
         bl_line_ids = range(bl_line_data["Attributes"][()].shape[0])
         names = np.vectorize(convert_ras_hdf_string)(
             bl_line_data["Attributes"][()]["Name"]
         )
-        geoms = list()
-        for pnt_start, pnt_cnt, part_start, part_cnt in bl_line_data["Polyline Info"][
-            ()
-        ]:
-            points = bl_line_data["Polyline Points"][()][
-                pnt_start : pnt_start + pnt_cnt
-            ]
-            if part_cnt == 1:
-                geoms.append(LineString(points))
-            else:
-                parts = bl_line_data["Polyline Parts"][()][
-                    part_start : part_start + part_cnt
-                ]
-                geoms.append(
-                    MultiLineString(
-                        list(
-                            points[part_pnt_start : part_pnt_start + part_pnt_cnt]
-                            for part_pnt_start, part_pnt_cnt in parts
-                        )
-                    )
-                )
+        geoms = self._get_polylines(self.BREAKLINES_PATH)
         return GeoDataFrame(
             {"bl_id": bl_line_ids, "name": names, "geometry": geoms},
             geometry="geometry",
@@ -400,36 +397,21 @@ def structures(self, datetime_to_str: bool = False) -> GeoDataFrame:
         GeoDataFrame
             A GeoDataFrame containing the model structures if they exist.
         """
-        if "/Geometry/Structures" not in self:
+        if self.GEOM_STRUCTURES_PATH not in self:
             return GeoDataFrame()
-        struct_data = self["/Geometry/Structures"]
+        struct_data = self[self.GEOM_STRUCTURES_PATH]
         v_conv_val = np.vectorize(convert_ras_hdf_value)
         sd_attrs = struct_data["Attributes"][()]
         struct_dict = {"struct_id": range(sd_attrs.shape[0])}
         struct_dict.update(
             {name: v_conv_val(sd_attrs[name]) for name in sd_attrs.dtype.names}
         )
-        geoms = list()
-        for pnt_start, pnt_cnt, part_start, part_cnt in struct_data["Centerline Info"][
-            ()
-        ]:
-            points = struct_data["Centerline Points"][()][
-                pnt_start : pnt_start + pnt_cnt
-            ]
-            if part_cnt == 1:
-                geoms.append(LineString(points))
-            else:
-                parts = struct_data["Centerline Parts"][()][
-                    part_start : part_start + part_cnt
-                ]
-                geoms.append(
-                    MultiLineString(
-                        list(
-                            points[part_pnt_start : part_pnt_start + part_pnt_cnt]
-                            for part_pnt_start, part_pnt_cnt in parts
-                        )
-                    )
-                )
+        geoms = self._get_polylines(
+            self.GEOM_STRUCTURES_PATH,
+            info_name="Centerline Info",
+            parts_name="Centerline Parts",
+            points_name="Centerline Points",
+        )
         struct_gdf = GeoDataFrame(
             struct_dict,
             geometry=geoms,
@@ -447,11 +429,101 @@ def connections(self) -> GeoDataFrame:  # noqa D102
     def ic_points(self) -> GeoDataFrame:  # noqa D102
         raise NotImplementedError
 
-    def reference_lines(self) -> GeoDataFrame:  # noqa D102
-        raise NotImplementedError
+    def reference_lines_names(
+        self, mesh_name: Optional[str] = None
+    ) -> Union[Dict[str, List[str]], List[str]]:
+        """Return reference line names.
 
-    def reference_points(self) -> GeoDataFrame:  # noqa D102
-        raise NotImplementedError
+        If a mesh name is provided, return a list of the reference line names for that mesh area.
+        If no mesh name is provided, return a dictionary of mesh names and their reference line names.
+
+        Parameters
+        ----------
+        mesh_name : str, optional
+            The name of the mesh area for which to return reference line names.
+
+        Returns
+        -------
+        Union[Dict[str, List[str]], List[str]]
+            A dictionary of mesh names and their reference line names if mesh_name is None.
+            A list of reference line names for the specified mesh area if mesh_name is not None.
+        """
+        attributes_path = f"{self.REFERENCE_LINES_PATH}/Attributes"
+        if mesh_name is None and attributes_path not in self:
+            return {m: [] for m in self.mesh_area_names()}
+        if mesh_name is not None and attributes_path not in self:
+            return []
+        attributes = self[attributes_path][()]
+        v_conv_str = np.vectorize(convert_ras_hdf_string)
+        names = np.vectorize(convert_ras_hdf_string)(attributes["Name"])
+        if mesh_name is not None:
+            return names[v_conv_str(attributes["SA-2D"]) == mesh_name].tolist()
+        mesh_names = np.vectorize(convert_ras_hdf_string)(attributes["SA-2D"])
+        return {m: names[mesh_names == m].tolist() for m in np.unique(mesh_names)}
+
+    def reference_lines(self) -> GeoDataFrame:
+        """Return the reference lines geometry and attributes.
+
+        Returns
+        -------
+        GeoDataFrame
+            A GeoDataFrame containing the reference lines if they exist.
+        """
+        attributes_path = f"{self.REFERENCE_LINES_PATH}/Attributes"
+        if attributes_path not in self:
+            return GeoDataFrame()
+        attributes = self[attributes_path][()]
+        refline_ids = range(attributes.shape[0])
+        v_conv_str = np.vectorize(convert_ras_hdf_string)
+        names = v_conv_str(attributes["Name"])
+        mesh_names = v_conv_str(attributes["SA-2D"])
+        try:
+            types = v_conv_str(attributes["Type"])
+        except ValueError:
+            # "Type" field doesn't exist -- observed in some RAS HDF files
+            types = np.array([""] * attributes.shape[0])
+        geoms = self._get_polylines(self.REFERENCE_LINES_PATH)
+        return GeoDataFrame(
+            {
+                "refln_id": refline_ids,
+                "refln_name": names,
+                "mesh_name": mesh_names,
+                "type": types,
+                "geometry": geoms,
+            },
+            geometry="geometry",
+            crs=self.projection(),
+        )
+
+    def reference_points(self) -> GeoDataFrame:
+        """Return the reference points geometry and attributes.
+
+        Returns
+        -------
+        GeoDataFrame
+            A GeoDataFrame containing the reference points if they exist.
+        """
+        attributes_path = f"{self.REFERENCE_POINTS_PATH}/Attributes"
+        if attributes_path not in self:
+            return GeoDataFrame()
+        ref_points_group = self[self.REFERENCE_POINTS_PATH]
+        attributes = ref_points_group["Attributes"][:]
+        v_conv_str = np.vectorize(convert_ras_hdf_string)
+        names = v_conv_str(attributes["Name"])
+        mesh_names = v_conv_str(attributes["SA/2D"])
+        cell_id = attributes["Cell Index"]
+        points = ref_points_group["Points"][()]
+        return GeoDataFrame(
+            {
+                "refpt_id": range(attributes.shape[0]),
+                "refpt_name": names,
+                "mesh_name": mesh_names,
+                "cell_id": cell_id,
+                "geometry": list(map(Point, points)),
+            },
+            geometry="geometry",
+            crs=self.projection(),
+        )
 
     def pump_stations(self) -> GeoDataFrame:  # noqa D102
         raise NotImplementedError