RASMapper Bank Lines¶
Python
# PIP MODE: use the installed ras-commander package.
# Uncomment if needed in a clean environment:
# !pip install --upgrade ras-commander pywin32 psutil comtypes h5py
import time
from pathlib import Path
import h5py
import pandas as pd
from ras_commander import RasExamples, init_ras_project
from ras_commander.gui import RasMapperBankLineWorkflow
Development Mode¶
When running from a local checkout instead of an installed package, set PYTHONPATH to the repository root before launching Jupyter. This notebook must run in an interactive Windows desktop session because it drives Ras.exe and RAS Mapper GUI menus.
RAS Mapper Bank Line Generation¶
This example opens a real HEC-RAS project, runs RAS Mapper's Create Bank Lines from XS Bank Stations command through the ras-commander GUI workflow, saves the geometry, and verifies that the geometry HDF contains generated river bank line polylines.
Python
RUN_GUI_UPDATE = True
PROJECT_NAME = "Balde Eagle Creek"
PROJECT_SUFFIX = "124_rasmapper_bank_lines"
GEOM_NUMBER = "01"
RAS_VERSION = "7.0"
TIMEOUT_SECONDS = 900
def find_repo_root():
for candidate in [Path.cwd(), Path.cwd().parent]:
if (candidate / "ras_commander").exists() and (candidate / "examples").exists():
return candidate
return Path.cwd()
REPO_ROOT = find_repo_root()
WORKING_ROOT = REPO_ROOT / "working" / "notebook_runs" / "124_rasmapper_bank_lines"
PROJECT_OUTPUT_ROOT = WORKING_ROOT / "example_projects"
PROJECT_OUTPUT_ROOT.mkdir(parents=True, exist_ok=True)
print(f"RUN_GUI_UPDATE = {RUN_GUI_UPDATE}")
print(f"Repo root: {REPO_ROOT}")
print(f"Working root: {WORKING_ROOT}")
Python
def close_hecras_processes(reason="cleanup"):
try:
import psutil
except Exception as exc:
print(f"HEC-RAS cleanup skipped; psutil unavailable: {exc}")
return 0
targets = []
for proc in psutil.process_iter(["pid", "name"]):
name = (proc.info.get("name") or "").lower()
if name in {"ras.exe", "rasmapper.exe"}:
targets.append(proc)
for proc in targets:
try:
proc.kill()
except Exception:
pass
if targets:
psutil.wait_procs(targets, timeout=5)
print(f"Closed {len(targets)} HEC-RAS/RASMapper process(es) during {reason}.")
return len(targets)
def geometry_record(ras_object, geom_number):
geom_number = str(geom_number).zfill(2)
geom_df = ras_object.geom_df.copy()
mask = geom_df["geom_number"].astype(str).str.zfill(2) == geom_number
matches = geom_df[mask]
if matches.empty:
raise ValueError(f"Geometry {geom_number} was not found in geom_df")
return matches.iloc[0]
def read_text_with_fallback(path):
path = Path(path)
for encoding in ("utf-8", "cp1252", "latin-1"):
try:
return path.read_text(encoding=encoding), encoding
except UnicodeDecodeError:
continue
return path.read_text(encoding="latin-1", errors="replace"), "latin-1-replace"
def bank_line_hdf_summary(hdf_path):
hdf_path = Path(hdf_path)
summary = {
"hdf_path": str(hdf_path),
"hdf_exists": hdf_path.exists(),
"river_bank_lines_group": False,
"polyline_count": 0,
"point_count": 0,
"datasets": "",
"mtime": hdf_path.stat().st_mtime if hdf_path.exists() else None,
}
if not hdf_path.exists():
return summary
with h5py.File(hdf_path, "r") as hdf:
group = hdf.get("Geometry/River Bank Lines")
if group is None:
return summary
summary["river_bank_lines_group"] = True
summary["datasets"] = ", ".join(sorted(group.keys()))
if "Polyline Info" in group:
summary["polyline_count"] = int(len(group["Polyline Info"]))
if "Polyline Points" in group:
summary["point_count"] = int(len(group["Polyline Points"]))
return summary
def geometry_text_summary(geometry_path):
geometry_path = Path(geometry_path)
text, encoding = read_text_with_fallback(geometry_path)
markers = ["Bank Sta=", "River Reach=", "Type RM Length L Ch R ="]
return {
"geometry_path": str(geometry_path),
"geometry_exists": geometry_path.exists(),
"geometry_encoding": encoding,
"geometry_mtime": geometry_path.stat().st_mtime,
"geometry_size_bytes": geometry_path.stat().st_size,
"text_markers_found": ", ".join(marker for marker in markers if marker in text),
}
Python
project_folder = RasExamples.extract_project(
PROJECT_NAME,
output_path=PROJECT_OUTPUT_ROOT,
suffix=PROJECT_SUFFIX,
)
test_ras = init_ras_project(project_folder, RAS_VERSION)
record = geometry_record(test_ras, GEOM_NUMBER)
geometry_path = Path(record["full_path"])
geometry_hdf_path = Path(record["hdf_path"])
before_summary = bank_line_hdf_summary(geometry_hdf_path)
display(pd.DataFrame([before_summary]))
Python
if RUN_GUI_UPDATE:
close_hecras_processes("pre-operation cleanup")
started_at = time.time()
result = RasMapperBankLineWorkflow.create_bank_lines_from_xs_bank_stations(
GEOM_NUMBER,
ras_object=test_ras,
timeout=TIMEOUT_SECONDS,
)
print(f"Workflow success: {result.success}")
if result.error:
raise RuntimeError(result.error)
if not result.success:
raise RuntimeError("RAS Mapper bank-line generation workflow failed")
else:
started_at = None
result = None
print("Dry run only. Set RUN_GUI_UPDATE = True to drive RAS Mapper.")
close_hecras_processes("post-operation cleanup")
Python
after_summary = bank_line_hdf_summary(geometry_hdf_path)
text_summary = geometry_text_summary(geometry_path)
validation = {
"workflow_success": None if result is None else result.success,
"before_polyline_count": before_summary["polyline_count"],
"after_polyline_count": after_summary["polyline_count"],
"after_point_count": after_summary["point_count"],
"bank_lines_generated": after_summary["polyline_count"] > before_summary["polyline_count"],
"geometry_updated_after_start": (
started_at is None or text_summary["geometry_mtime"] >= started_at - 2
),
"hdf_updated_after_start": (
started_at is None
or after_summary["mtime"] is not None
and after_summary["mtime"] >= started_at - 2
),
**text_summary,
"geometry_hdf_path": after_summary["hdf_path"],
"bank_line_hdf_datasets": after_summary["datasets"],
}
display(pd.DataFrame([validation]))
if RUN_GUI_UPDATE:
assert validation["workflow_success"] is True
assert validation["bank_lines_generated"] is True
assert validation["after_point_count"] > 0