Merge pull request #25 from fema-ffrd/feature/stac-funcs

Feature/stac funcs

src/rashdf/base.py

@@ -1,4 +1,6 @@
 import h5py
+from .utils import hdf5_attrs_to_dict
+from typing import Dict
 
 
 class RasHdf(h5py.File):
@@ -42,7 +44,35 @@ def open_uri(
         --------
         >>> results_hdf = RasHdf.open_uri("s3://my-bucket/results.hdf")
         """
-        import fsspec  # type: ignore
+        import fsspec
 
         remote_file = fsspec.open(uri, mode="rb", **fsspec_kwargs)
         return cls(remote_file.open(), **h5py_kwargs)
+
+    def get_attrs(self, attr_path: str) -> Dict:
+        """Convert attributes from a HEC-RAS HDF file into a Python dictionary for a given attribute path.
+
+        Parameters
+        ----------
+            attr_path (str): The path within the HEC-RAS HDF file where the desired attributes are located (Ex. "Plan Data/Plan Parameters").
+
+        Returns
+        -------
+            plan_attrs (dict): Dictionary filled with attributes at given path, if attributes exist at that path.
+        """
+        attr_object = self.get(attr_path)
+
+        if attr_object:
+            return hdf5_attrs_to_dict(attr_object.attrs)
+
+        return {}
+
+    def get_root_attrs(self):
+        """Returns attributes at root level of HEC-RAS HDF file.
+
+        Returns
+        -------
+        dict
+            Dictionary filled with HEC-RAS HDF root attributes.
+        """
+        return self.get_attrs("/")

src/rashdf/geom.py

@@ -1,5 +1,5 @@
 from .base import RasHdf
-from .utils import convert_ras_hdf_string
+from .utils import convert_ras_hdf_string, get_first_hdf_group, hdf5_attrs_to_dict
 
 import numpy as np
 from geopandas import GeoDataFrame
@@ -17,6 +17,13 @@
 
 
 class RasGeomHdf(RasHdf):
+    GEOM_PATH = "Geometry"
+    GEOM_STRUCTURES_PATH = f"{GEOM_PATH}/Structures"
+    FLOW_AREA_2D_PATH = f"{GEOM_PATH}/2D Flow Areas"
+
+    def __init__(self, name: str, **kwargs):
+        super().__init__(name, **kwargs)
+
     def projection(self) -> Optional[CRS]:
         """Return the projection of the RAS geometry as a
         pyproj.CRS object.
@@ -193,6 +200,45 @@ def mesh_cell_faces(self) -> GeoDataFrame:
                 face_dict["geometry"].append(LineString(coordinates))
         return GeoDataFrame(face_dict, geometry="geometry", crs=self.projection())
 
+    def get_geom_attrs(self):
+        """Returns base geometry attributes from a HEC-RAS HDF file.
+
+        Returns
+        -------
+        dict
+            Dictionary filled with base geometry attributes.
+        """
+        return self.get_attrs(self.GEOM_PATH)
+
+    def get_geom_structures_attrs(self):
+        """Returns geometry structures attributes from a HEC-RAS HDF file.
+
+        Returns
+        -------
+        dict
+            Dictionary filled with geometry structures attributes.
+        """
+        return self.get_attrs(self.GEOM_STRUCTURES_PATH)
+
+    def get_geom_2d_flow_area_attrs(self):
+        """Returns geometry 2d flow area attributes from a HEC-RAS HDF file.
+
+        Returns
+        -------
+        dict
+            Dictionary filled with geometry 2d flow area attributes.
+        """
+        try:
+            d2_flow_area = get_first_hdf_group(self.get(self.FLOW_AREA_2D_PATH))
+        except AttributeError:
+            raise AttributeError(
+                f"Unable to get 2D Flow Area; {self.FLOW_AREA_2D_PATH} group not found in HDF5 file."
+            )
+
+        d2_flow_area_attrs = hdf5_attrs_to_dict(d2_flow_area.attrs)
+
+        return d2_flow_area_attrs
+
     def bc_lines(self) -> GeoDataFrame:
         """Return the 2D mesh area boundary condition lines.
 

src/rashdf/plan.py

@@ -1,8 +1,78 @@
 from .geom import RasGeomHdf
-
+from typing import Dict
 from geopandas import GeoDataFrame
 
 
 class RasPlanHdf(RasGeomHdf):
+    PLAN_INFO_PATH = "Plan Data/Plan Information"
+    PLAN_PARAMS_PATH = "Plan Data/Plan Parameters"
+    PRECIP_PATH = "Event Conditions/Meteorology/Precipitation"
+    RESULTS_UNSTEADY_PATH = "Results/Unsteady"
+    RESULTS_UNSTEADY_SUMMARY_PATH = f"{RESULTS_UNSTEADY_PATH}/Summary"
+    VOLUME_ACCOUNTING_PATH = f"{RESULTS_UNSTEADY_PATH}/Volume Accounting"
+
+    def __init__(self, name: str, **kwargs):
+        super().__init__(name, **kwargs)
+
+    def get_plan_info_attrs(self) -> Dict:
+        """Returns plan information attributes from a HEC-RAS HDF plan file.
+
+        Returns
+        -------
+        dict
+            Dictionary filled with plan information attributes.
+        """
+        return self.get_attrs(self.PLAN_INFO_PATH)
+
+    def get_plan_param_attrs(self) -> Dict:
+        """Returns plan parameter attributes from a HEC-RAS HDF plan file.
+
+        Returns
+        -------
+        dict
+            Dictionary filled with plan parameter attributes.
+        """
+        return self.get_attrs(self.PLAN_PARAMS_PATH)
+
+    def get_meteorology_precip_attrs(self) -> Dict:
+        """Returns precipitation attributes from a HEC-RAS HDF plan file.
+
+        Returns
+        -------
+        dict
+            Dictionary filled with precipitation attributes.
+        """
+        return self.get_attrs(self.PRECIP_PATH)
+
+    def get_results_unsteady_attrs(self) -> Dict:
+        """Returns unsteady attributes from a HEC-RAS HDF plan file.
+
+        Returns
+        -------
+        dict
+            Dictionary filled with unsteady attributes.
+        """
+        return self.get_attrs(self.RESULTS_UNSTEADY_PATH)
+
+    def get_results_unsteady_summary_attrs(self) -> Dict:
+        """Returns results unsteady summary attributes from a HEC-RAS HDF plan file.
+
+        Returns
+        -------
+        dict
+            Dictionary filled with results summary attributes.
+        """
+        return self.get_attrs(self.RESULTS_UNSTEADY_SUMMARY_PATH)
+
+    def get_results_volume_accounting_attrs(self) -> Dict:
+        """Returns volume accounting attributes from a HEC-RAS HDF plan file.
+
+        Returns
+        -------
+        dict
+            Dictionary filled with volume accounting attributes.
+        """
+        return self.get_attrs(self.VOLUME_ACCOUNTING_PATH)
+
     def enroachment_points(self) -> GeoDataFrame:
         raise NotImplementedError

src/rashdf/utils.py

@@ -1,6 +1,6 @@
 import numpy as np
-
-from typing import Any, List, Tuple, Union
+import h5py
+from typing import Any, List, Tuple, Union, Optional
 
 from datetime import datetime, timedelta
 import re
@@ -168,3 +168,48 @@ def convert_ras_hdf_value(
     # Convert all other types to string
     else:
         return str(value)
+
+
+def hdf5_attrs_to_dict(attrs: dict, prefix: str = None) -> dict:
+    """
+    Convert a dictionary of attributes from an HDF5 file into a Python dictionary.
+
+    Parameters:
+    ----------
+        attrs (dict): The attributes to be converted.
+        prefix (str, optional): An optional prefix to prepend to the keys.
+
+    Returns:
+    ----------
+        dict: A dictionary with the converted attributes.
+    """
+    results = {}
+    for k, v in attrs.items():
+        value = convert_ras_hdf_value(v)
+        if prefix:
+            key = f"{prefix}:{k}"
+        else:
+            key = k
+        results[key] = value
+    return results
+
+
+def get_first_hdf_group(parent_group: h5py.Group) -> Optional[h5py.Group]:
+    """
+    Get the first HDF5 group from a parent group.
+
+    This function iterates over the items in the parent group and returns the first item that is an instance of
+     h5py.Group. If no such item is found, it returns None.
+
+    Parameters:
+    ----------
+        parent_group (h5py.Group): The parent group to search in.
+
+    Returns:
+    ----------
+        Optional[h5py.Group]: The first HDF5 group in the parent group, or None if no group is found.
+    """
+    for _, item in parent_group.items():
+        if isinstance(item, h5py.Group):
+            return item
+    return None

tests/test_geom.py

@@ -71,3 +71,42 @@ def test_refinement_regions():
     rr_json = TEST_JSON / "refinement_regions.json"
     with RasGeomHdf(MUNCIE_G05) as ghdf:
         assert _gdf_matches_json(ghdf.refinement_regions(), rr_json)
+
+
+def test_get_geom_attrs(tmp_path):
+    attrs_to_set = {"test_attribute1": "test_str1", "test_attribute2": 500}
+
+    with h5py.File(tmp_path / "test.hdf", "w") as f:
+        geom_group = f.create_group(RasGeomHdf.GEOM_PATH)
+        for key, value in attrs_to_set.items():
+            geom_group.attrs[key] = value
+
+    ras_hdf = RasGeomHdf(tmp_path / "test.hdf")
+
+    assert ras_hdf.get_geom_attrs() == attrs_to_set
+
+
+def test_get_geom_structures_attrs(tmp_path):
+    attrs_to_set = {"test_attribute1": "test_str1", "test_attribute2": 500}
+
+    with h5py.File(tmp_path / "test.hdf", "w") as f:
+        structures_group = f.create_group(RasGeomHdf.GEOM_STRUCTURES_PATH)
+        for key, value in attrs_to_set.items():
+            structures_group.attrs[key] = value
+
+    ras_hdf = RasGeomHdf(tmp_path / "test.hdf")
+
+    assert ras_hdf.get_geom_structures_attrs() == attrs_to_set
+
+
+def test_get_geom_2d_flow_area_attrs(tmp_path):
+    attrs_to_set = {"test_attribute1": "test_str1", "test_attribute2": 500}
+
+    with h5py.File(tmp_path / "test.hdf", "w") as f:
+        flow_area_group = f.create_group(f"{RasGeomHdf.FLOW_AREA_2D_PATH}/group")
+        for key, value in attrs_to_set.items():
+            flow_area_group.attrs[key] = value
+
+    ras_hdf = RasGeomHdf(tmp_path / "test.hdf")
+
+    assert ras_hdf.get_geom_2d_flow_area_attrs() == attrs_to_set

tests/test_plan.py

@@ -0,0 +1,71 @@
+from src.rashdf import RasPlanHdf
+
+import h5py
+
+attrs_to_set = {"test_attribute1": "test_str1", "test_attribute2": 500}
+
+
+def test_get_plan_info_attrs(tmp_path):
+    with h5py.File(tmp_path / "test.hdf", "w") as f:
+        group = f.create_group(RasPlanHdf.PLAN_INFO_PATH)
+        for key, value in attrs_to_set.items():
+            group.attrs[key] = value
+
+    ras_plan_hdf = RasPlanHdf(tmp_path / "test.hdf")
+
+    assert ras_plan_hdf.get_plan_info_attrs() == attrs_to_set
+
+
+def test_get_plan_param_attrs(tmp_path):
+    with h5py.File(tmp_path / "test.hdf", "w") as f:
+        group = f.create_group(RasPlanHdf.PLAN_PARAMS_PATH)
+        for key, value in attrs_to_set.items():
+            group.attrs[key] = value
+
+    ras_plan_hdf = RasPlanHdf(tmp_path / "test.hdf")
+
+    assert ras_plan_hdf.get_plan_param_attrs() == attrs_to_set
+
+
+def test_get_meteorology_precip_attrs(tmp_path):
+    with h5py.File(tmp_path / "test.hdf", "w") as f:
+        group = f.create_group(RasPlanHdf.PRECIP_PATH)
+        for key, value in attrs_to_set.items():
+            group.attrs[key] = value
+
+    ras_plan_hdf = RasPlanHdf(tmp_path / "test.hdf")
+
+    assert ras_plan_hdf.get_meteorology_precip_attrs() == attrs_to_set
+
+
+def test_get_results_unsteady_attrs(tmp_path):
+    with h5py.File(tmp_path / "test.hdf", "w") as f:
+        group = f.create_group(RasPlanHdf.RESULTS_UNSTEADY_PATH)
+        for key, value in attrs_to_set.items():
+            group.attrs[key] = value
+
+    ras_plan_hdf = RasPlanHdf(tmp_path / "test.hdf")
+
+    assert ras_plan_hdf.get_results_unsteady_attrs() == attrs_to_set
+
+
+def test_get_results_unsteady_summary_attrs(tmp_path):
+    with h5py.File(tmp_path / "test.hdf", "w") as f:
+        group = f.create_group(RasPlanHdf.RESULTS_UNSTEADY_SUMMARY_PATH)
+        for key, value in attrs_to_set.items():
+            group.attrs[key] = value
+
+    ras_plan_hdf = RasPlanHdf(tmp_path / "test.hdf")
+
+    assert ras_plan_hdf.get_results_unsteady_summary_attrs() == attrs_to_set
+
+
+def test_get_results_volume_accounting_attrs(tmp_path):
+    with h5py.File(tmp_path / "test.hdf", "w") as f:
+        group = f.create_group(RasPlanHdf.VOLUME_ACCOUNTING_PATH)
+        for key, value in attrs_to_set.items():
+            group.attrs[key] = value
+
+    ras_plan_hdf = RasPlanHdf(tmp_path / "test.hdf")
+
+    assert ras_plan_hdf.get_results_volume_accounting_attrs() == attrs_to_set