Dam Breach Results¶
Overview¶
This notebook demonstrates extracting dam breach analysis results from HEC-RAS HDF5 files. Breach modeling simulates catastrophic failure of dams or levees.
Breach Analysis Components: - Breach Formation: Time-dependent widening and deepening of breach - Breach Outflow: Flow through developing breach opening - Downstream Propagation: Flood wave routing downstream - Reservoir Drawdown: Water surface in reservoir during breach
HDF5 Structure Used by ras-commander¶
/Results/Unsteady/Output/Output Blocks/Base Output/Unsteady Time Series/
└── SA 2D Area Conn/
└── {structure}/
├── Structure Variables # Flow and water-surface time series
└── Breaching Variables # Time-varying breach geometry
Breach Parameters (time-varying): - Width: Breach widens over time (linear, sine, or user-defined) - Invert: Breach deepens (erodes downward) - Flow: Combination of weir and orifice flow - Formation Time: Time from initiation to full formation
Breach Modeling Theory¶
Failure Modes: - Overtopping: Erosion starts at crest, progresses downstream - Piping: Internal erosion, sudden collapse - Earthquake: Instantaneous breach - Programmed: User-defined breach progression
Flow Equations:
Qweir = Cw * Ltop * H^1.5 (Weir flow - unsubmerged)
Qorifice = Co * Abottom * sqrt(H) (Orifice flow - submerged)
Qtotal = Qweir + Qorifice (Combined)
Where: - Cw = weir coefficient - Ltop = top width of breach - H = head difference - Co = orifice coefficient - Abottom = bottom area of breach
Reference Documentation¶
- HEC-RAS User's Manual, Chapter 16: Dam Break Analysis
- FEMA Dam Safety Guidelines - Breach modeling requirements
- ASDSO Breach Parameters - Recommended breach parameter selection
Regulatory Context¶
Dam breach analysis typically required for: - Hazard Classification: Determine dam hazard potential (High, Significant, Low) - Emergency Action Plans (EAP): Inundation mapping for evacuation planning - Dam Safety Inspections: Assess consequences of failure - FEMA Flood Maps: Breach scenarios for floodplain mapping
# Import ras-commander from the active Python environment.
from ras_commander import HdfResultsBreach, HdfResultsPlan, HdfStruc, RasBreach, RasCmdr, RasExamples, RasPlan, init_ras_project, ras
# Standard imports
import pandas as pd
import matplotlib.pyplot as plt
import numpy as np
import shutil
# Verify which version loaded
import ras_commander
print(f"✓ Loaded: {ras_commander.__file__}")
Parameters¶
Configure these values to customize the notebook for your project.
# =============================================================================
# PARAMETERS - Edit these to customize the notebook
# =============================================================================
from pathlib import Path
# Project Configuration
PROJECT_NAME = "BaldEagleCrkMulti2D" # Example project to extract
RAS_VERSION = None # Optional override, for example "7.0"
# HDF Analysis Settings
PLAN = "19" # Plan number (for HDF file path)
print(f"Configuration: {PROJECT_NAME} project, Plan {PLAN}, RAS override={RAS_VERSION}")
Dam Breach Results Extraction and Sensitivity Analysis¶
This notebook demonstrates: 1. Extracting baseline breach results from HDF files 2. Reading breach parameters from plan files 3. Modifying parameters iteratively (one parameter at a time) 4. Comparing results across different scenarios 5. Visualizing sensitivity to parameter changes
The project, template plan, and HEC-RAS version are configured in the Parameters cell above.
Workflow: - Extract baseline results - Clone plan and modify one parameter - Re-extract results and compare - Repeat for multiple parameters - Plot all scenarios together
Setup and Imports¶
1. Extract and Initialize Project¶
# Extract the BaldEagleCrkMulti2D example project using static method
example_project_folder = RasExamples.extract_project(PROJECT_NAME, suffix="16")
print(f"Extracted project to: {example_project_folder}")
# Verify the path exists
print(f"BaldEagleCrkMulti2D project exists: {example_project_folder.exists()}")
# Set project_path variable for compatibility with rest of notebook
project_path = example_project_folder
# Initialize the project
init_ras_project(project_path, RAS_VERSION)
effective_ras_version = ras.ras_version
print(f"\nInitialized project: {ras.project_name}")
print(f"Effective HEC-RAS execution version: {effective_ras_version}")
print(f"\nAvailable plans:")
ras.plan_df
# This is the SA to 2D Dam Break Run
template_plan = PLAN
2. BASELINE: Extract Existing Results from the Template Plan¶
First, compute and analyze the baseline breach behavior from the configured template plan (PLAN).
Important Note: HDF Results Files¶
This example project may not include pre-computed HDF results. These files are generated when HEC-RAS runs a simulation.
The next section computes the configured template plan through RasCmdr.compute_plan(), then reads its generated HDF. All execution occurs in the notebook's extracted disposable project copy.
# Initialize variables to None (prevents NameError in later cells)
target_structure = None
hdf_target_structure = None
baseline_ts = pd.DataFrame()
baseline_summary = pd.DataFrame()
baseline_params = None
baseline_geom = []
scenarios = {}
summaries = {}
scenario_1_plan = None
scenario_2_plan = None
2.1 Identify Breach Structures¶
# List breach structures from PLAN FILE (for parameter operations)
# This ensures structure names match between listing and read_breach_block()
breach_structures_list = RasBreach.list_breach_structures_plan(template_plan)
print("Breach Structures in Plan File:")
for struct in breach_structures_list:
if struct['structure']: # Filter empty names
status = f"stored Breach Loc is_active={struct['is_active']}"
location = f"{struct['river']}/{struct['reach']}/RS {struct['station']}" if struct['river'] else "No location"
print(f" - {struct['structure']}: {status} ({location})")
# Get active structure names for parameter operations
breach_structures = [s['structure'] for s in breach_structures_list
if s['structure'] and s['is_active']]
if breach_structures:
target_structure = breach_structures[0]
print(f"\nTarget structure for analysis: {target_structure}")
print(f" This name will work with RasBreach.read_breach_block()")
else:
print("\nWARNING: No structure has stored Breach Loc is_active=True.")
target_structure = None
if target_structure:
# Read breach parameters from plan file
baseline_params = RasBreach.read_breach_block(template_plan, target_structure)
print(f"Baseline Parameters for {target_structure}:")
print("=" * 80)
print(f"\nStored Breach Loc is_active: {baseline_params['is_active']}")
print(f"\nKey Parameter Values:")
for key in ['Breach Method', 'Breach Geom', 'Breach Start', 'Breach Progression']:
if key in baseline_params['values']:
print(f" {key}: {baseline_params['values'][key]}")
# Parse geometry values for modification
geom_str = baseline_params['values'].get('Breach Geom', '')
baseline_geom = [x.strip() for x in geom_str.split(',')]
print(f"\nBaseline Geometry (parsed): {baseline_geom}")
# Explain Breach Geom field structure
if len(baseline_geom) >= 9:
print("\nBreach Geom Stored Record (CSV, 9 required fields; field 9 is optional in legacy nine-field records):")
print(f" [0] Centerline/Station: {baseline_geom[0]} ft")
print(f" [1] Final Bottom Width: {baseline_geom[1]} ft")
print(f" [2] Final Bottom Elevation: {baseline_geom[2]} ft <-- KEY PARAMETER")
print(f" [3] Left Side Slope: {baseline_geom[3]} (H:V)")
print(f" [4] Right Side Slope: {baseline_geom[4]} (H:V)")
print(f" [5] Stored Failure-Mode Flag: {baseline_geom[5]}")
print(f" [6] Piping Coefficient: {baseline_geom[6]}")
print(f" [7] Initial Piping Elevation: {baseline_geom[7]} ft")
print(f" [8] Breach Formation Time: {baseline_geom[8]} hrs")
if len(baseline_geom) >= 10:
print(f" [9] Breach Weir Coefficient: {baseline_geom[9]}")
else:
print(" [9] Breach Weir Coefficient: not stored in this record")
print("\n--- Example: Update Final Bottom Elevation ---")
print(f"Current value: {baseline_geom[2]} ft")
print("To change to 605 ft:")
print(" example_plan = RasPlan.clone_plan(template_plan, 'Raised breach bottom')")
print(" new_geom = baseline_geom.copy()")
print(" new_geom[2] = 605")
print(" RasBreach.update_breach_block(example_plan, target_structure, geom_values=new_geom)")
else:
print("Skipping parameter reading - no breach structure available")
baseline_params = None
baseline_geom = []
# First, compute the template plan to generate HDF results
print(f"Computing plan {template_plan} to generate HDF results...")
print("(This may take 1-2 minutes)")
RasCmdr.compute_plan(template_plan, skip_existing=True, num_cores=2)
print(f"Plan {template_plan} complete")
# Resolve the HDF structure name separately from the plan-file structure name.
hdf_breach_info = HdfStruc.get_sa2d_breach_info(template_plan)
hdf_breach_structures = hdf_breach_info.loc[
hdf_breach_info['has_breach'], 'structure'
].tolist()
if target_structure in hdf_breach_structures:
hdf_target_structure = target_structure
elif len(hdf_breach_structures) == 1:
hdf_target_structure = hdf_breach_structures[0]
else:
hdf_target_structure = None
if len(hdf_breach_structures) > 1:
print("Ambiguous HDF breach structures; no automatic fallback selected.")
# Now extract breach results from the HDF.
if hdf_target_structure:
# Extract complete breach time series from HDF results
baseline_ts = HdfResultsBreach.get_breach_timeseries(template_plan, hdf_target_structure)
print(f"Baseline Time Series Extracted: {baseline_ts.shape}")
print(f"\nColumns: {list(baseline_ts.columns)}")
print(f"\nFirst few timesteps:")
print(baseline_ts.head())
# Get summary statistics from HDF results
baseline_summary = HdfResultsBreach.get_breach_summary(template_plan, hdf_target_structure)
print(f"\nBaseline Summary Statistics:")
print(baseline_summary.to_string(index=False))
# Store only nonempty baseline results for later comparison.
scenarios = (
{f'Baseline (Plan {template_plan})': baseline_ts.copy()}
if not baseline_ts.empty
else {}
)
summaries = (
{f'Baseline (Plan {template_plan})': baseline_summary.copy()}
if not baseline_summary.empty
else {}
)
else:
print("Skipping baseline extraction - no HDF breach structure available")
baseline_ts = pd.DataFrame()
baseline_summary = pd.DataFrame()
scenarios = {}
summaries = {}
Easy Parameter Modification with set_breach_geom()¶
NEW FUNCTION: RasBreach.set_breach_geom() provides a clean interface for modifying individual breach parameters without manually parsing/reconstructing the CSV.
# Example: Update just Final Bottom Elevation (most common modification)
if target_structure:
print("Example: Update Final Bottom Elevation to 605 ft")
print("=" * 60)
print("\nClone the baseline first, then use set_breach_geom():")
print(" example_plan = RasPlan.clone_plan(template_plan, 'Raised breach bottom')")
print(" RasBreach.set_breach_geom(example_plan, target_structure,")
print(" final_bottom_elev=605)")
print("\nThis automatically:")
print(" 1. Reads current Breach Geom values")
print(" 2. Updates ONLY the final_bottom_elev field (index 2)")
print(" 3. Preserves all other parameters")
print(" 4. Writes back to plan file with backup")
print("\n\nOther common modifications:")
print("\n# Increase breach width by 50%")
print(" current_width = 200 # Read the final bottom width from baseline_params")
print(" RasBreach.set_breach_geom(example_plan, target_structure,")
print(" final_bottom_width=current_width * 1.5)")
print("\n# Change formation time")
print(" RasBreach.set_breach_geom(example_plan, target_structure,")
print(" formation_time=3.5)")
print("\n# Update multiple parameters at once")
print(" RasBreach.set_breach_geom(example_plan, target_structure,")
print(" final_bottom_elev=605,")
print(" final_bottom_width=250,")
print(" formation_time=3.0)")
else:
print("No target structure available for examples")
3. SCENARIO ANALYSIS: Modify Parameters and Compare Results¶
Now we'll create multiple scenarios by modifying breach parameters one at a time.
Workflow for each scenario:
1. Clone the configured template plan
2. Modify one parameter in the clone
3. Compute the cloned plans in parallel through RasCmdr.compute_parallel()
4. Extract the generated HDF results
5. Compare each scenario with the baseline
Scenario 1: Increase Breach Width by 50%¶
if target_structure and baseline_geom and len(baseline_geom) >= 2:
# Clone plan
scenario_1_plan = RasPlan.clone_plan(template_plan, "Scenario 1: +50% Width")
# Modify the final breach bottom width (field 1).
original_width = float(baseline_geom[1])
new_width = original_width * 1.5
print(f"\nModifying breach width:")
print(f" Original: {original_width} ft")
print(f" New: {new_width} ft (+50%)")
# Update the plan
RasBreach.set_breach_geom(
scenario_1_plan,
target_structure,
final_bottom_width=new_width,
)
print(f"\n✓ Scenario 1 plan created: {scenario_1_plan}")
print(f" Ready for parallel HEC-RAS execution: plan {scenario_1_plan}")
else:
print("Skipping Scenario 1 - insufficient baseline data")
Scenario 2: Decrease Breach Formation Time by 50%¶
if target_structure and baseline_geom and len(baseline_geom) >= 9:
# Clone plan
scenario_2_plan = RasPlan.clone_plan(template_plan, "Scenario 2: -50% Formation Time")
# Formation time is field 8; use the named setter to avoid index errors
original_time = float(baseline_geom[8])
new_time = original_time * 0.5
print(f"\nModifying breach formation time:")
print(f" Original: {original_time} hrs")
print(f" New: {new_time} hrs (-50%)")
# Update the plan
RasBreach.set_breach_geom(
scenario_2_plan,
target_structure,
formation_time=new_time
)
print(f"\n✓ Scenario 2 plan created: {scenario_2_plan}")
print(f" Ready for parallel HEC-RAS execution: plan {scenario_2_plan}")
else:
print("Skipping Scenario 2 - insufficient baseline data")
scenario_plans_to_compute = [
plan for plan in (scenario_1_plan, scenario_2_plan) if plan is not None
]
parallel_computed_folder = example_project_folder.parent / f"{example_project_folder.name}_parallelcomputed"
if scenario_plans_to_compute:
RasCmdr.compute_parallel(
scenario_plans_to_compute,
max_workers=4,
num_cores=2,
dest_folder=Path(parallel_computed_folder),
overwrite_dest=True,
)
# Re-initialize explicitly with the configured HEC-RAS version.
init_ras_project(parallel_computed_folder, effective_ras_version)
print(f"Reinitialized results with HEC-RAS version: {ras.ras_version}")
results_project_path = Path(parallel_computed_folder)
else:
results_project_path = project_path
print("Skipping parallel execution: no scenario plans were created.")
# Display results summary from results_df after parallel execution
# This shows execution status, timing, and any errors/warnings for each plan
ras.results_df[['plan_number', 'plan_title', 'completed', 'has_errors', 'has_warnings', 'runtime_complete_process_hours']]
# Scenario definitions
scenario_plans = {
name: plan
for name, plan in {
'Scenario 1: +50% Width': scenario_1_plan,
'Scenario 2: -50% Formation Time': scenario_2_plan,
}.items()
if plan is not None
}
# Extract results for each computed scenario from HDF files.
if hdf_target_structure:
for scenario_name, plan_num in scenario_plans.items():
ts = HdfResultsBreach.get_breach_timeseries(plan_num, hdf_target_structure)
summary = HdfResultsBreach.get_breach_summary(plan_num, hdf_target_structure)
if not ts.empty:
scenarios[scenario_name] = ts
print(f"✓ Extracted time series: {scenario_name}")
else:
print(f"⚠ No breach time series found for: {scenario_name}")
if not summary.empty:
summaries[scenario_name] = summary
print(f"✓ Extracted summary: {scenario_name}")
else:
print(f"⚠ No breach summary found for: {scenario_name}")
else:
print("Skipping scenario extraction - no HDF breach structure available.")
print(f"\nTotal scenarios with results: {len(scenarios)}")
4. Extract Results from All Scenarios¶
The preceding parallel execution generated HDF files for the cloned scenario plans. This section inspects their computation messages and structured breach results.
import textwrap
# Get the raw computation messages string from the results HDF for scenario 1.
comp_msgs = (
HdfResultsPlan.get_compute_messages(scenario_1_plan)
if scenario_1_plan is not None
else ""
)
def pretty_print_compute_messages(msg: str) -> None:
"""
Nicely format and print RAS compute messages. Strips unnecessary escapes,
ensures readable blocks, and optionally highlights warnings.
"""
if not msg:
print("No computation messages found.")
return
# Replace carriage returns, unify newlines
msg = msg.replace('\r\n', '\n').replace('\r', '\n')
# Collapse excessive blank lines to at most 2
lines = msg.split('\n')
pretty_lines = []
blank_count = 0
for line in lines:
if line.strip() == '':
blank_count += 1
if blank_count <= 2:
pretty_lines.append('')
else:
blank_count = 0
# Optionally add highlighting for warnings/errors
l_strip = line.lstrip()
if l_strip.lower().startswith("warning") or "error" in l_strip.lower():
pretty_lines.append("⚠️ " + line)
else:
pretty_lines.append(line)
# Optionally wrap long lines for readability
final_lines = []
for l in pretty_lines:
if len(l) > 120:
final_lines.extend(textwrap.wrap(l, width=120))
else:
final_lines.append(l)
# Print result
print('\n'.join(final_lines))
pretty_print_compute_messages(comp_msgs)
# List SA/2D connection structures in HDF results
hdf_structures = HdfStruc.list_sa2d_connections(template_plan)
print("SA/2D Connection Structures in HDF:")
for struct in hdf_structures:
print(f" - {struct}")
# Get breach capability information
breach_info = HdfStruc.get_sa2d_breach_info(template_plan)
print("\nBreach Capability Information:")
print(breach_info.to_string(index=False))
# Get list of structures with breach capability
breach_structures = breach_info[breach_info['has_breach']]['structure'].tolist()
print(f"\nStructures with breach capability: {breach_structures}")
# Keep the HDF name separate from the plan-file target used by RasBreach.
if target_structure in breach_structures:
hdf_target_structure = target_structure
elif len(breach_structures) == 1:
hdf_target_structure = breach_structures[0]
else:
hdf_target_structure = None
if len(breach_structures) > 1:
print("Ambiguous HDF breach structures; no automatic fallback selected.")
print(f"\nPlan-file target structure: {target_structure}")
print(f"HDF target structure: {hdf_target_structure}")
# Get breach-specific variables (width, depth, slopes over time)
breach_vars = (
HdfResultsBreach.get_breaching_variables(template_plan, hdf_target_structure)
if hdf_target_structure
else pd.DataFrame()
)
breach_vars
5. Compare Results Across All Scenarios¶
Visualize all scenarios together to understand parameter sensitivity.
5.1 Flow Hydrograph Comparison¶
flow_scenarios = {
name: frame for name, frame in scenarios.items()
if not frame.empty and {'datetime', 'total_flow'}.issubset(frame.columns)
}
colors = ['blue', 'red', 'green', 'orange', 'purple']
linestyles = ['-', '--', '-.', ':', '-']
if flow_scenarios:
fig, ax = plt.subplots(figsize=(14, 6))
for idx, (scenario_name, ts_data) in enumerate(flow_scenarios.items()):
color = colors[idx % len(colors)]
linestyle = linestyles[idx % len(linestyles)]
ax.plot(ts_data['datetime'], ts_data['total_flow'],
label=scenario_name, color=color, linestyle=linestyle, linewidth=2)
ax.set_xlabel('Time', fontsize=12)
ax.set_ylabel('Total Flow (cfs)', fontsize=12)
ax.set_title(f'{hdf_target_structure} - Flow Hydrograph Comparison',
fontsize=14, fontweight='bold')
ax.legend(loc='best', fontsize=10)
ax.grid(True, alpha=0.3)
plt.xticks(rotation=45)
plt.tight_layout()
plt.show()
else:
print("No nonempty scenario time series contain datetime and total_flow.")
5.2 Peak Flow Comparison (Bar Chart)¶
peak_summaries = {
name: frame for name, frame in summaries.items()
if not frame.empty and 'max_total_flow' in frame.columns
}
if peak_summaries:
# Extract peak flows
scenario_names = list(peak_summaries.keys())
peak_flows = [peak_summaries[name].iloc[0]['max_total_flow']
for name in scenario_names]
# Create bar chart
fig, ax = plt.subplots(figsize=(12, 6))
bars = ax.bar(range(len(scenario_names)), peak_flows,
color=['blue', 'red', 'green', 'orange', 'purple'][:len(scenario_names)])
ax.set_xticks(range(len(scenario_names)))
ax.set_xticklabels(scenario_names, rotation=45, ha='right')
ax.set_ylabel('Peak Flow (cfs)', fontsize=12)
ax.set_title(f'{hdf_target_structure} - Peak Flow Comparison',
fontsize=14, fontweight='bold')
ax.grid(True, axis='y', alpha=0.3)
# Add value labels on bars
for bar, value in zip(bars, peak_flows):
height = bar.get_height()
ax.text(bar.get_x() + bar.get_width()/2., height,
f'{value:.0f}',
ha='center', va='bottom', fontsize=10, fontweight='bold')
plt.tight_layout()
plt.show()
# Print percent differences from baseline
if len(peak_flows) > 1:
baseline_flow = peak_flows[0]
print("\nPeak Flow Differences from Baseline:")
print("=" * 60)
for i, (name, flow) in enumerate(zip(scenario_names, peak_flows)):
if i == 0:
print(f"{name}: {flow:.0f} cfs (baseline)")
else:
diff_pct = ((flow - baseline_flow) / baseline_flow) * 100
print(f"{name}: {flow:.0f} cfs ({diff_pct:+.1f}%)")
else:
print("No nonempty summaries contain max_total_flow.")
5.3 Breach Width Evolution Comparison¶
width_scenarios = {
name: frame for name, frame in scenarios.items()
if (not frame.empty
and {'datetime', 'bottom_width'}.issubset(frame.columns)
and frame['bottom_width'].notna().any())
}
if width_scenarios:
if width_scenarios:
fig, ax = plt.subplots(figsize=(14, 6))
for idx, (scenario_name, ts_data) in enumerate(width_scenarios.items()):
if ts_data['bottom_width'].notna().any():
color = colors[idx % len(colors)]
linestyle = linestyles[idx % len(linestyles)]
ax.plot(ts_data['datetime'], ts_data['bottom_width'],
label=scenario_name, color=color, linestyle=linestyle,
linewidth=2, marker='o', markersize=4)
ax.set_xlabel('Time', fontsize=12)
ax.set_ylabel('Breach Width (ft)', fontsize=12)
ax.set_title(f'{hdf_target_structure} - Breach Width Evolution Comparison',
fontsize=14, fontweight='bold')
ax.legend(loc='best', fontsize=10)
ax.grid(True, alpha=0.3)
plt.xticks(rotation=45)
plt.tight_layout()
plt.show()
else:
print("No breach width data available (breach may not have formed)")
else:
print("No nonempty scenario time series contain breach width data.")
5.4 Summary Table Comparison¶
valid_summaries = {
name: frame for name, frame in summaries.items() if not frame.empty
}
if valid_summaries:
# Combine all summaries into a comparison table
comparison_data = []
for scenario_name, summary_df in valid_summaries.items():
row = summary_df.iloc[0].to_dict()
row['Scenario'] = scenario_name
comparison_data.append(row)
comparison_df = pd.DataFrame(comparison_data)
# Select key columns for display
display_cols = ['Scenario', 'max_total_flow', 'max_breach_flow',
'final_breach_width', 'final_breach_depth',
'max_hw', 'max_tw']
# Filter to available columns
display_cols = [col for col in display_cols if col in comparison_df.columns]
print("\nScenario Comparison Summary:")
print("=" * 100)
print(comparison_df[display_cols].to_string(index=False))
# Export to CSV
output_file = results_project_path / "breach_scenario_comparison.csv"
comparison_df.to_csv(output_file, index=False)
print(f"\nComparison table exported to: {output_file}")
else:
print("No nonempty summary data available for comparison table.")
5.5 Comprehensive Multi-Panel Comparison¶
dashboard_keys = [
name for name, frame in scenarios.items()
if (name in summaries
and not frame.empty
and not summaries[name].empty
and {'datetime', 'total_flow', 'hw', 'tw'}.issubset(frame.columns)
and 'max_total_flow' in summaries[name].columns)
]
if len(dashboard_keys) > 1:
# Create 2x2 subplot grid
fig, axes = plt.subplots(2, 2, figsize=(16, 12))
# Plot 1: Flow comparison
for idx, scenario_name in enumerate(dashboard_keys):
ts_data = scenarios[scenario_name]
color = colors[idx % len(colors)]
axes[0, 0].plot(ts_data['datetime'], ts_data['total_flow'],
label=scenario_name, color=color, linewidth=2)
axes[0, 0].set_ylabel('Total Flow (cfs)', fontsize=11)
axes[0, 0].set_title('Flow Hydrographs', fontsize=12, fontweight='bold')
axes[0, 0].legend(fontsize=8)
axes[0, 0].grid(True, alpha=0.3)
# Plot 2: Peak flows bar chart
scenario_names = dashboard_keys
peak_flows = [summaries[name].iloc[0]['max_total_flow'] for name in scenario_names]
bars = axes[0, 1].bar(range(len(scenario_names)), peak_flows,
color=colors[:len(scenario_names)])
axes[0, 1].set_xticks(range(len(scenario_names)))
axes[0, 1].set_xticklabels(scenario_names, rotation=30, ha='right')
axes[0, 1].set_ylabel('Peak Flow (cfs)', fontsize=11)
axes[0, 1].set_title('Peak Flow Comparison', fontsize=12, fontweight='bold')
axes[0, 1].grid(True, axis='y', alpha=0.3)
# Plot 3: HW/TW for baseline
baseline_ts = scenarios[dashboard_keys[0]]
axes[1, 0].plot(baseline_ts['datetime'], baseline_ts['hw'],
label='HW', color='blue', linewidth=2)
axes[1, 0].plot(baseline_ts['datetime'], baseline_ts['tw'],
label='TW', color='red', linewidth=2)
axes[1, 0].set_xlabel('Time', fontsize=11)
axes[1, 0].set_ylabel('Elevation (ft)', fontsize=11)
axes[1, 0].set_title('Baseline Water Levels', fontsize=12, fontweight='bold')
axes[1, 0].legend()
axes[1, 0].grid(True, alpha=0.3)
# Plot 4: Breach width comparison (if available)
has_width = False
for idx, scenario_name in enumerate(dashboard_keys):
ts_data = scenarios[scenario_name]
if 'bottom_width' in ts_data.columns and ts_data['bottom_width'].notna().any():
color = colors[idx % len(colors)]
axes[1, 1].plot(ts_data['datetime'], ts_data['bottom_width'],
label=scenario_name, color=color, linewidth=2, marker='o')
has_width = True
if has_width:
axes[1, 1].set_xlabel('Time', fontsize=11)
axes[1, 1].set_ylabel('Breach Width (ft)', fontsize=11)
axes[1, 1].set_title('Breach Width Evolution', fontsize=12, fontweight='bold')
axes[1, 1].legend(fontsize=8)
axes[1, 1].grid(True, alpha=0.3)
else:
axes[1, 1].text(0.5, 0.5, 'No Breach Width Data',
ha='center', va='center', fontsize=14,
transform=axes[1, 1].transAxes)
fig.suptitle(f'{hdf_target_structure} - Breach Scenario Analysis Dashboard',
fontsize=16, fontweight='bold', y=0.995)
plt.tight_layout()
plt.show()
else:
print("Need at least two scenarios with compatible time series and summaries.")
6. Export All Results¶
if scenarios:
# Export each scenario's time series
for scenario_name, ts_data in scenarios.items():
# Create safe filename
safe_name = scenario_name.replace(' ', '_').replace(':', '').replace('+', 'plus')
filename = results_project_path / f"breach_{safe_name}.csv"
ts_data.to_csv(filename, index=False)
print(f"Exported: {filename.name}")
print(f"\nAll scenario data exported to: {results_project_path}")
else:
print("No scenario data to export")
Summary¶
This notebook demonstrated:
✅ Baseline Analysis: - Extracted existing breach results from HDF - Read baseline breach parameters from plan file - Visualized baseline behavior
✅ Scenario Creation: - Cloned plans to create new scenarios - Modified breach parameters one at a time: - Scenario 1: Increased breach width by 50% - Scenario 2: Decreased formation time by 50%
✅ Results Comparison: - Extracted results from all scenarios - Compared flow hydrographs - Compared peak flows - Compared breach geometry evolution - Created comprehensive comparison dashboard
✅ Data Export: - Exported time series for all scenarios - Exported comparison summary table
Key Functions Used:¶
# HDF Results Extraction (use HdfResultsBreach and HdfStruc)
HdfStruc.list_sa2d_connections(plan) # List structures in HDF
HdfStruc.get_sa2d_breach_info(plan) # Get breach capability info
HdfResultsBreach.get_breach_timeseries(plan, structure) # Extract time series
HdfResultsBreach.get_breach_summary(plan, structure) # Extract summary stats
HdfResultsBreach.get_breaching_variables(plan, structure) # Breach geometry evolution
HdfResultsBreach.get_structure_variables(plan, structure) # Structure flow variables
# Plan File Parameter Management (use RasBreach)
RasBreach.list_breach_structures_plan(plan) # List structures in plan file
RasBreach.read_breach_block(plan, structure) # Read parameters
RasBreach.update_breach_block(plan, structure, geom_values=[...]) # Modify parameters
Architectural Pattern:¶
ras-commander separates HDF and plain text operations: - RasBreach → Breach PARAMETERS in plan files (.p##) - HdfResultsBreach → Breach RESULTS from HDF files (.p##.hdf) - HdfStruc → Structure listings and metadata from HDF
Important: Use plan file methods for parameter operations to ensure structure names match!
Breach Geom Field Structure:¶
# Breach Geom CSV format: fields 0-8 are required; field 9 is version-dependent.
[0] Centerline/Station # ft
[1] Final Bottom Width # ft
[2] Final Bottom Elevation # ft <-- Example: change this to 605
[3] Left Side Slope # H:V ratio
[4] Right Side Slope # H:V ratio
[5] Stored Failure-Mode Flag # raw True/False value
[6] Piping Coefficient # dimensionless
[7] Initial Piping Elevation # ft
[8] Breach Formation Time # hrs
[9] Breach Weir Coefficient # optional/version-dependent
# Example: clone the baseline, then update Final Bottom Elevation
example_plan = RasPlan.clone_plan(template_plan, "Raised breach bottom")
new_geom = baseline_geom.copy()
new_geom[2] = 605 # Set to 605 ft
RasBreach.update_breach_block(example_plan, target_structure, geom_values=new_geom)
7. Advanced Breach Features (NEW)¶
This section demonstrates new RasBreach features:
create_breach_block()- Create new breach blocks from scratch- DLBreach (Method 9) - Physics-based erosion modeling with soil properties
- Advanced Parameters - User growth ratio, mass wasting options
- BreachBlock Values - Inspect DLBreach parameters returned by the public reader
DLBreach Overview¶
DLBreach (Dam-breach Lake Breach) is Method 9 in HEC-RAS, providing physics-based erosion modeling that uses soil properties to simulate breach progression more realistically than parametric methods.
Key DLBreach Parameters:
- dlb_methods: 7-value list of method flags (first value is primary method 0-9)
- dlb_soil_type: Soil type index (0-7: Sand, Loamy Sand, Sandy Loam, Loam, Silt Loam, Sandy Clay Loam, Clay Loam, Clay)
- dlb_soil_properties: 7 erosion parameters (critical shear stress, erodibility, etc.)
- dlb_breach_direction: Breach direction (0=downstream, 1=upstream, 2=both)
7.1 Creating New Breach Blocks¶
Use RasBreach.create_breach_block() to add breach capability to structures that don't already have it.
# Demonstrate create_breach_block() - creates new breach for a structure
# NOTE: This is demonstration code only - requires a structure without existing breach
print("create_breach_block() - Add breach capability to a structure")
print("=" * 70)
print("""
# Syntax:
result = RasBreach.create_breach_block(
plan_input="01", # Plan number or path
structure_name="NewDam", # Structure name (must not already have breach)
river="Main River", # Optional: River name for 1D location
reach="Upper Reach", # Optional: Reach name
station="12345.0", # Optional: River station
is_active=True, # Set the stored Breach Loc flag
create_backup=True # Create backup before modifying
)
# Returns dict with created block details:
# {
# 'structure_name': 'NewDam',
# 'is_active': True,
# 'values': {'Breach Method': ' 0', 'Breach Geom': '...', ...}
# }
# After creation, customize parameters:
RasBreach.update_breach_block(
"01", "NewDam",
geom_values=[5700, 200, 605, 0.5, 0.5, True, 0.5, 630, 2, 2.6],
method=0 # User-entered breach geometry
)
""")
print("\nCreated blocks start with Method 0 (user-entered data) defaults.")
print("Use update_breach_block() to configure specific breach parameters.")
7.2 DLBreach (Method 9) Configuration¶
DLBreach is HEC-RAS's physics-based erosion model. It uses soil properties to simulate breach progression more realistically than parametric methods.
Soil Types (dlb_soil_type): | Index | Soil Type | |-------|-----------| | 0 | Sand | | 1 | Loamy Sand | | 2 | Sandy Loam | | 3 | Loam | | 4 | Silt Loam | | 5 | Sandy Clay Loam | | 6 | Clay Loam | | 7 | Clay |
Soil Properties (dlb_soil_properties): 7-value list 1. Critical shear stress (Pa) 2. Erodibility coefficient (m³/N·s) 3. Angle of repose (degrees) 4. Porosity 5. Unit weight (N/m³) 6. Cohesion (Pa) 7. Friction angle (degrees)
# Configure DLBreach (Method 9) for physics-based erosion
print("DLBreach Configuration Example")
print("=" * 70)
print("""
# Configure existing breach block for DLBreach (Method 9)
RasBreach.update_breach_block(
plan_input="01",
structure_name="Dam",
# Method 9 = DLBreach
method=9,
# DLBreach-specific parameters
dlb_methods=[9, 0, 0, 0, 0, 0, 0], # Primary method in first position
dlb_soil_type=2, # Sandy Loam
dlb_soil_properties=[ # Custom soil properties
1.5, # Critical shear stress (Pa)
0.001, # Erodibility coefficient (m³/N·s)
35.0, # Angle of repose (degrees)
0.35, # Porosity
18000, # Unit weight (N/m³)
5000, # Cohesion (Pa)
30.0 # Friction angle (degrees)
],
dlb_breach_direction=0, # 0=downstream, 1=upstream, 2=both
# Standard geometry parameters still apply
geom_values=[5700, 200, 605, 0.5, 0.5, True, 0.5, 630, 2, 2.6]
)
""")
print("DLBreach models breach erosion using soil mechanics principles.")
print("More realistic than parametric methods for detailed dam safety studies.")
7.3 BreachBlock Values¶
read_breach_block() returns a public dictionary containing raw values and parsed table_rows. Read DLBreach and advanced fields from block["values"], as shown below. The nested BreachBlock parser type has convenience getters, but the public reader intentionally returns the stable dictionary shape rather than that internal object.
# Demonstrate reading breach parameters including DLBreach settings
if target_structure:
# Read the breach block
block = RasBreach.read_breach_block(template_plan, target_structure)
print(f"Breach Block for '{target_structure}':")
print("=" * 70)
print(f"\nMethod: {block['values'].get('Breach Method', 'Not set')}")
print(f"Stored Breach Loc is_active: {block['is_active']}")
# Show key parameters
print("\nKey Parameters:")
for key in ['Breach Geom', 'Breach Start', 'Breach Progression']:
if key in block['values']:
value = block['values'][key]
if len(str(value)) > 60:
value = str(value)[:60] + "..."
print(f" {key}: {value}")
# Show advanced parameters if present
print("\nAdvanced Parameters (if configured):")
advanced_keys = [
'Mass Wasting Options',
'Breach Use User Defined Growth Ratio',
'Breach User Defined Growth Ratio',
'DLBreach Methods',
'DLBreach SoilType',
'DLBreach Soil Properties',
'DLBreach Breach Direction'
]
for key in advanced_keys:
if key in block['values']:
print(f" {key}: {block['values'][key]}")
else:
print("No breach structure available for demonstration")
7.4 Advanced Features Summary¶
New RasBreach Methods:
# Create new breach block
RasBreach.create_breach_block(plan, structure_name, river="", reach="", station="")
# Update with DLBreach (Method 9)
RasBreach.update_breach_block(plan, structure,
method=9,
dlb_methods=[9,0,0,0,0,0,0],
dlb_soil_type=2,
dlb_soil_properties=[1.5, 0.001, 35.0, 0.35, 18000, 5000, 30.0],
dlb_breach_direction=0
)
# Advanced parameters
RasBreach.update_breach_block(plan, structure,
mass_wasting_option=1,
user_growth_flag=1,
user_growth_ratio=1.5
)
Use Cases: - TnTech Dam Breach Dashboard: Full breach parameter control via API - Automated Sensitivity Analysis: Programmatic DLBreach configuration - Batch Processing: Create/configure breach blocks across multiple plans