ras2cng.geometry

ras2cng.geometry

Geometry export functions for ras2cng.

Supports: - HEC-RAS geometry HDF: .g??.hdf (mesh, BCs, structures, XS, centerlines) - HEC-RAS text geometry: .g?? (XS cut lines, centerlines, storage areas)

export_geometry_layers(geom_input, output, layer=None)

Export HEC-RAS geometry layers to GeoParquet.

Parameters:

Name	Type	Description	Default
geom_input	Path	Path to .g?? or .g??.hdf	required
output	Path	Output GeoParquet path	required
layer	Optional[str]	Layer name (mesh_cells, cross_sections, centerlines, bc_lines, breaklines, refinement_regions, reference_lines, reference_points, structures, mesh_areas, storage_areas). None = auto-select best.	None

Source code in ras2cng/geometry.py

def export_geometry_layers(
    geom_input: Path,
    output: Path,
    layer: Optional[str] = None,
):
    """Export HEC-RAS geometry layers to GeoParquet.

    Args:
        geom_input: Path to *.g?? or *.g??.hdf
        output: Output GeoParquet path
        layer: Layer name (mesh_cells, cross_sections, centerlines, bc_lines,
               breaklines, refinement_regions, reference_lines, reference_points,
               structures, mesh_areas, storage_areas). None = auto-select best.
    """
    geom_path = Path(geom_input)

    if _is_hdf_geometry(geom_path):
        export_hdf_geometry(geom_path, output, layer=layer)
    elif _is_text_geometry(geom_path):
        export_text_geometry(geom_path, output, layer=layer)
    else:
        raise ValueError(
            f"Unsupported geometry file format: {geom_path.name} (suffixes={geom_path.suffixes})"
        )

export_hdf_geometry(hdf_path, output, layer=None)

Export geometry from a HDF geometry file (*.g??.hdf).

When layer is None, prefers mesh_cells then falls back to first available.

Source code in ras2cng/geometry.py

def export_hdf_geometry(hdf_path: Path, output: Path, layer: Optional[str] = None):
    """Export geometry from a HDF geometry file (*.g??.hdf).

    When layer is None, prefers mesh_cells then falls back to first available.
    """
    if layer is not None:
        gdf = _extract_hdf_layer(hdf_path, layer)
        if gdf is None:
            raise ValueError(f"Layer '{layer}' could not be extracted from {hdf_path.name}")
        output.parent.mkdir(parents=True, exist_ok=True)
        _prepare_for_parquet(gdf).to_parquet(output, compression="snappy", index=False)
        return

    # Auto-select: try all layers, prefer mesh_cells
    layers = {}
    for lname in ALL_HDF_LAYERS:
        gdf = _extract_hdf_layer(hdf_path, lname)
        if gdf is not None:
            layers[lname] = gdf

    if not layers:
        raise ValueError("No geometry layers could be extracted from HDF geometry file")

    preferred = "mesh_cells" if "mesh_cells" in layers else next(iter(layers))
    output.parent.mkdir(parents=True, exist_ok=True)
    _prepare_for_parquet(layers[preferred]).to_parquet(output, compression="snappy", index=False)

export_all_hdf_layers(hdf_path, output_dir, skip_empty=True)

Export all available geometry layers from a single HDF file.

Parameters:

Name	Type	Description	Default
hdf_path	Path	Path to *.g??.hdf geometry file	required
output_dir	Path	Directory to write individual layer parquet files	required
skip_empty	bool	If True, silently skip layers that return no data	True

Returns:

Type	Description
dict[str, Path]	Dict of {layer_name: parquet_path} for successfully written layers

Source code in ras2cng/geometry.py

def export_all_hdf_layers(
    hdf_path: Path,
    output_dir: Path,
    skip_empty: bool = True,
) -> dict[str, Path]:
    """Export all available geometry layers from a single HDF file.

    Args:
        hdf_path: Path to *.g??.hdf geometry file
        output_dir: Directory to write individual layer parquet files
        skip_empty: If True, silently skip layers that return no data

    Returns:
        Dict of {layer_name: parquet_path} for successfully written layers
    """
    output_dir.mkdir(parents=True, exist_ok=True)
    written: dict[str, Path] = {}

    for layer_name in ALL_HDF_LAYERS:
        out_path = output_dir / f"{layer_name}.parquet"
        try:
            gdf = _extract_hdf_layer(hdf_path, layer_name)
        except Exception as e:
            print(f"Warning: Could not extract '{layer_name}' from {hdf_path.name}: {e}")
            continue
        if gdf is None:
            if not skip_empty:
                print(f"Info: Layer '{layer_name}' not available in {hdf_path.name}")
            continue
        _prepare_for_parquet(gdf).to_parquet(out_path, compression="snappy", index=False)
        written[layer_name] = out_path

    return written

export_text_geometry(geom_path, output, layer=None)

Export geometry from a plain text geometry file (*.g??).

Layers available from text files: cross_sections, centerlines, storage_areas

Source code in ras2cng/geometry.py

def export_text_geometry(geom_path: Path, output: Path, layer: Optional[str] = None):
    """Export geometry from a plain text geometry file (*.g??).

    Layers available from text files: cross_sections, centerlines, storage_areas
    """
    safe_path = _ensure_utf8_readable(geom_path)
    layers = {}

    if layer is None or layer == "cross_sections":
        try:
            gdf = GeomParser.get_xs_cut_lines(safe_path)
            if len(gdf) > 0:
                layers["cross_sections"] = gdf
        except Exception as e:
            print(f"Warning: Could not extract XS cut lines: {e}")

    if layer is None or layer == "centerlines":
        try:
            gdf = GeomParser.get_river_centerlines(safe_path)
            if len(gdf) > 0:
                layers["centerlines"] = gdf
        except Exception as e:
            print(f"Warning: Could not extract river centerlines: {e}")

    if layer is None or layer == "storage_areas":
        try:
            gdf = GeomStorage.get_storage_areas(safe_path)
            if len(gdf) > 0:
                layers["storage_areas"] = gdf
        except Exception as e:
            print(f"Warning: Could not extract storage areas: {e}")

    if not layers:
        raise ValueError("No geometry layers could be extracted from text geometry file")

    if layer is not None:
        if layer not in layers:
            raise ValueError(
                f"Layer '{layer}' not available. Available: {list(layers.keys())}"
            )
        output.parent.mkdir(parents=True, exist_ok=True)
        _prepare_for_parquet(layers[layer]).to_parquet(output, compression="snappy", index=False)
        return

    preferred = "cross_sections" if "cross_sections" in layers else next(iter(layers))
    output.parent.mkdir(parents=True, exist_ok=True)
    _prepare_for_parquet(layers[preferred]).to_parquet(output, compression="snappy", index=False)

export_all_text_layers(geom_path, output_dir)

Export all available geometry layers from a text geometry file.

Parameters:

Name	Type	Description	Default
geom_path	Path	Path to *.g?? text geometry file	required
output_dir	Path	Directory to write individual layer parquet files	required

Returns:

Type	Description
dict[str, Path]	Dict of {layer_name: parquet_path} for successfully written layers

Source code in ras2cng/geometry.py

def export_all_text_layers(
    geom_path: Path,
    output_dir: Path,
) -> dict[str, Path]:
    """Export all available geometry layers from a text geometry file.

    Args:
        geom_path: Path to *.g?? text geometry file
        output_dir: Directory to write individual layer parquet files

    Returns:
        Dict of {layer_name: parquet_path} for successfully written layers
    """
    safe_path = _ensure_utf8_readable(geom_path)
    output_dir.mkdir(parents=True, exist_ok=True)
    written: dict[str, Path] = {}

    extractors = {
        "cross_sections": lambda p: GeomParser.get_xs_cut_lines(p),
        "centerlines":    lambda p: GeomParser.get_river_centerlines(p),
        "storage_areas":  lambda p: GeomStorage.get_storage_areas(p),
    }

    for layer_name, extractor in extractors.items():
        try:
            gdf = extractor(safe_path)
            if len(gdf) == 0:
                continue
            out_path = output_dir / f"{layer_name}.parquet"
            _prepare_for_parquet(gdf).to_parquet(out_path, compression="snappy", index=False)
            written[layer_name] = out_path
        except Exception as e:
            print(f"Warning: Could not extract '{layer_name}' from {geom_path.name}: {e}")

    return written

merge_all_layers(hdf_path=None, text_path=None, *, sort=True)

Extract and merge all geometry layers into a single GeoDataFrame.

HDF layers use their base names (e.g. mesh_cells). Text layers get a _text suffix (e.g. cross_sections_text). All layers are distinguished by a layer column.

Parameters:

Name	Type	Description	Default
hdf_path	Optional[Path]	Path to *.g??.hdf geometry file (or None)	None
text_path	Optional[Path]	Path to *.g?? text geometry file (or None)	None
sort	bool	If True, apply Hilbert spatial sort within each layer	True

Returns:

Type	Description
Optional[GeoDataFrame]	A merged GeoDataFrame with layer column, or None if nothing extracted

Source code in ras2cng/geometry.py

def merge_all_layers(
    hdf_path: Optional[Path] = None,
    text_path: Optional[Path] = None,
    *,
    sort: bool = True,
) -> Optional[gpd.GeoDataFrame]:
    """Extract and merge all geometry layers into a single GeoDataFrame.

    HDF layers use their base names (e.g. ``mesh_cells``). Text layers get
    a ``_text`` suffix (e.g. ``cross_sections_text``). All layers are
    distinguished by a ``layer`` column.

    Args:
        hdf_path: Path to ``*.g??.hdf`` geometry file (or None)
        text_path: Path to ``*.g??`` text geometry file (or None)
        sort: If True, apply Hilbert spatial sort within each layer

    Returns:
        A merged GeoDataFrame with ``layer`` column, or None if nothing extracted
    """
    all_gdfs: list[gpd.GeoDataFrame] = []

    # --- HDF layers ---
    if hdf_path is not None:
        for layer_name in ALL_HDF_LAYERS:
            try:
                gdf = _extract_hdf_layer(hdf_path, layer_name)
            except Exception as e:
                print(f"Warning: Could not extract '{layer_name}' from {hdf_path.name}: {e}")
                continue
            if gdf is None or len(gdf) == 0:
                continue
            gdf = _prepare_for_parquet(gdf)
            gdf["layer"] = layer_name
            if sort:
                gdf = _hilbert_sort(gdf)
            all_gdfs.append(gdf)

    # --- Text layers ---
    if text_path is not None:
        safe_path = _ensure_utf8_readable(text_path)
        text_extractors = {
            "cross_sections": lambda p: GeomParser.get_xs_cut_lines(p),
            "centerlines":    lambda p: GeomParser.get_river_centerlines(p),
            "storage_areas":  lambda p: GeomStorage.get_storage_areas(p),
        }
        for layer_name, extractor in text_extractors.items():
            try:
                gdf = extractor(safe_path)
                if len(gdf) == 0:
                    continue
            except Exception as e:
                print(f"Warning: Could not extract '{layer_name}' from {text_path.name}: {e}")
                continue
            gdf = _prepare_for_parquet(gdf)
            gdf["layer"] = f"{layer_name}_text"
            if sort:
                gdf = _hilbert_sort(gdf)
            all_gdfs.append(gdf)

    if not all_gdfs:
        return None

    merged = pd.concat(all_gdfs, ignore_index=True)
    return gpd.GeoDataFrame(merged, geometry="geometry")