diff --git a/src/rashdf/base.py b/src/rashdf/base.py index c13b9cc..15f3ddf 100644 --- a/src/rashdf/base.py +++ b/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("/") diff --git a/src/rashdf/geom.py b/src/rashdf/geom.py index 8447342..c67624c 100644 --- a/src/rashdf/geom.py +++ b/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. diff --git a/src/rashdf/plan.py b/src/rashdf/plan.py index a64ec57..bd90cc6 100644 --- a/src/rashdf/plan.py +++ b/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 diff --git a/src/rashdf/utils.py b/src/rashdf/utils.py index 0d6dc81..7eeae60 100644 --- a/src/rashdf/utils.py +++ b/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 diff --git a/tests/test_geom.py b/tests/test_geom.py index 8673727..204d60d 100644 --- a/tests/test_geom.py +++ b/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 diff --git a/tests/test_plan.py b/tests/test_plan.py new file mode 100644 index 0000000..fe98f21 --- /dev/null +++ b/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