OpenFOAM Job Submission Example - DAPI

This example demonstrates how to submit and monitor an OpenFOAM CFD simulation using dapi. OpenFOAM is a free, open-source computational fluid dynamics (CFD) software package.

Overview¶

This example covers the essential workflow for running OpenFOAM simulations:

Installing and importing dapi
Setting up OpenFOAM job parameters
Configuring solvers, mesh generation, and decomposition
Submitting and monitoring CFD jobs
Post-processing results with force coefficient analysis

Complete Example¶

Step 1: Install and Import dapi¶

# Install dapi package
!pip install dapi --user --quiet

# Import required modules
from dapi import DSClient
import json

What this does:

Installs the DesignSafe API package from PyPI
Imports the main client class and JSON for pretty-printing job requests

Step 2: Initialize Client¶

# Initialize DesignSafe client
ds = DSClient()

What this does:

Creates an authenticated connection to DesignSafe services
Handles OAuth2 authentication automatically
Sets up connections to Tapis API, file systems, and job services

Authentication: dapi supports multiple authentication methods including environment variables, .env files, and interactive prompts. For detailed authentication setup instructions, see the authentication guide.

Step 3: Configure Job Parameters¶

# Job configuration parameters
ds_path: str = "/MyData/template-notebooks/tapis3/OpenFOAM/DH1_run" # Path to OpenFOAM case directory
max_job_minutes: int = 10 # Maximum runtime in minutes
tacc_allocation: str = "ASC25049" # TACC allocation to charge
app_id_to_use = "openfoam-stampede3" # OpenFOAM application ID

# OpenFOAM-specific environment variables
openfoam_env_vars = [
 {"key": "mesh", "value": "On"}, # Enable mesh generation with blockMesh
 {"key": "solver", "value": "pisoFoam"}, # CFD solver to use
 {"key": "decomp", "value": "On"} # Enable domain decomposition for parallel runs
]

What each parameter does:

ds_path: DesignSafe path to your OpenFOAM case directory containing 0/, constant/, and system/ folders
max_job_minutes: Maximum wall-clock time for the job (prevents runaway simulations)
tacc_allocation: Your TACC allocation account (required for compute time billing)
app_id_to_use: Specific OpenFOAM application version on DesignSafe
openfoam_env_vars: OpenFOAM-specific configuration:
mesh: "On" - Runs blockMesh to generate computational mesh
solver: "pisoFoam" - Transient, incompressible Navier-Stokes solver
decomp: "On" - Enables parallel domain decomposition for multi-core runs

Alternative solver options:

# Different OpenFOAM solvers you can use
solvers = {
 "pisoFoam": "Transient, incompressible (general purpose)",
 "simpleFoam": "Steady-state, incompressible (RANS)",
 "pimpleFoam": "Transient, incompressible (large time steps)",
 "rhoSimpleFoam": "Steady-state, compressible",
 "sonicFoam": "Transient, compressible (high speed flows)"
}

Step 4: Convert Path to URI¶

# Convert DesignSafe path to Tapis URI format
input_uri = ds.files.to_uri(ds_path)
print(f"Input Directory Tapis URI: {input_uri}")

What this does:

Converts human-readable DesignSafe paths (like /MyData/...) to Tapis URI format
Tapis URIs are required for job submission and follow the pattern: tapis://system/path
Automatically detects your username and the correct storage system

Step 5: Generate Job Request¶

# Generate job request dictionary using app defaults
job_dict = ds.jobs.generate(
 app_id=app_id_to_use,
 input_dir_uri=input_uri,
 max_minutes=max_job_minutes,
 allocation=tacc_allocation,
 archive_system="designsafe",
 extra_env_vars=openfoam_env_vars,
 input_dir_param_name="Case Directory" # OpenFOAM apps use "Case Directory" instead of "Input Directory"
)
print(json.dumps(job_dict, indent=2, default=str))

What each parameter does:

app_id: Specifies which application to run
input_dir_uri: Location of your OpenFOAM case files
max_minutes: Job timeout (prevents infinite runs)
allocation: TACC account to charge for compute time
archive_system: Where to store results (“designsafe” = your MyData folder)
extra_env_vars: OpenFOAM-specific settings passed to the application
input_dir_param_name: OpenFOAM apps expect “Case Directory” not “Input Directory”

Additional options you can add:

# Extended job configuration options
job_dict = ds.jobs.generate(
 app_id=app_id_to_use,
 input_dir_uri=input_uri,
 max_minutes=max_job_minutes,
 allocation=tacc_allocation,
 
 # Resource configuration
 node_count=2, # Number of compute nodes
 cores_per_node=48, # Cores per node (max depends on system)
 memory_mb=96000, # Memory in MB per node
 queue="normal", # Queue: "development", "normal", "large", etc.
 
 # Job metadata
 job_name="my_cfd_simulation", # Custom job name
 description="Wind flow around building", # Job description
 tags=["research", "cfd", "wind-engineering"], # Searchable tags
 
 # Archive configuration
 archive_system="designsafe", # Where to store results
 archive_path="openfoam-results", # Custom archive subdirectory
 
 # Additional environment variables
 extra_env_vars=[
 {"key": "mesh", "value": "On"},
 {"key": "solver", "value": "pisoFoam"},
 {"key": "decomp", "value": "On"},
 {"key": "OMP_NUM_THREADS", "value": "4"} # OpenMP threads per MPI process
 ]
)

Step 6: Customize Resources¶

# Customize job settings (optional)
job_dict["nodeCount"] = 1 # Use single node
job_dict["coresPerNode"] = 2 # Use 2 cores for parallel simulation
print(json.dumps(job_dict, indent=2, default=str))

What this does:

Overrides default resource allocation from the app
nodeCount: Number of compute nodes (1 for small jobs, multiple for large simulations)
coresPerNode: CPU cores per node (enables parallel processing)
More cores = faster solution but higher cost

Resource guidelines:

# Resource selection guidelines
resources = {
 "small_case": {"nodes": 1, "cores": 2, "time": 30}, # < 100K cells
 "medium_case": {"nodes": 1, "cores": 16, "time": 120}, # 100K - 1M cells
 "large_case": {"nodes": 2, "cores": 48, "time": 480}, # > 1M cells
}

Step 7: Submit Job¶

# Submit the job to TACC
submitted_job = ds.jobs.submit(job_dict)
print(f"Job UUID: {submitted_job.uuid}")

What this does:

Sends the job request to TACC’s job scheduler
Returns a SubmittedJob object for monitoring
Job UUID is a unique identifier for tracking

Step 8: Monitor Job¶

# Monitor job execution until completion
final_status = submitted_job.monitor(interval=15) # Check every 15 seconds
print(f"Job {submitted_job.uuid} finished with status: {final_status}")

What this does:

Polls job status at specified intervals (15 seconds)
Shows progress bars for different job phases
Returns final status when job completes
interval=15 means check every 15 seconds (can be adjusted)

Job status meanings:

job_statuses = {
 "PENDING": "Job submitted but not yet processed",
 "PROCESSING_INPUTS": "Input files being staged",
 "QUEUED": "Job waiting in scheduler queue",
 "RUNNING": "Job actively executing",
 "ARCHIVING": "Output files being archived",
 "FINISHED": "Job completed successfully",
 "FAILED": "Job failed during execution"
}

Step 9: Check Results¶

# Interpret and display job outcome
ds.jobs.interpret_status(final_status, submitted_job.uuid)

# Display job runtime summary
submitted_job.print_runtime_summary(verbose=False)

# Get current job status
current_status = ds.jobs.status(submitted_job.uuid)
print(f"Current status: {current_status}")

# Display last status message from TACC
print(f"Last message: {submitted_job.last_message}")

What each command does:

interpret_status: Provides human-readable explanation of job outcome
print_runtime_summary: Shows time spent in each job phase (queued, running, etc.)
status: Gets current job status (useful for checking later)
last_message: Shows last status message from the job scheduler

Step 10: View Job Output¶

# Display job output from stdout
stdout_content = submitted_job.get_output_content("tapisjob.out", max_lines=50)
if stdout_content:
 print("Job output:")
 print(stdout_content)

What this does:

tapisjob.out contains all console output from your OpenFOAM simulation
max_lines=50 limits output to last 50 lines (prevents overwhelming output)
Shows OpenFOAM solver progress, residuals, and timing information

Step 11: Access Results¶

# List contents of job archive directory
archive_uri = submitted_job.archive_uri
print(f"Archive URI: {archive_uri}")
outputs = ds.files.list(archive_uri)
for item in outputs:
 print(f"- {item.name} ({item.type})")

What this does:

archive_uri: Location where job results are stored
ds.files.list: Lists all files and directories in the archive
Shows output files like mesh, solution fields, and post-processing data

Typical OpenFOAM output files:

typical_outputs = {
 "inputDirectory/": "Copy of your case directory with results",
 "tapisjob.out": "Console output from OpenFOAM",
 "tapisjob.err": "Error messages (if any)",
 "tapisjob.sh": "Job script that was executed",
 "postProcessing/": "Force coefficients, residuals, monitoring data",
 "processor*/": "Parallel decomposed solution (if using multiple cores)"
}

Post-processing Results¶

Extract Force Coefficients¶

# Convert archive URI to local path for analysis
archive_path = ds.files.to_path(archive_uri)
print(f"Archive path: {archive_path}")

# Import plotting libraries
import numpy as np
import matplotlib.pyplot as plt
import os

# Load force coefficient data using pandas
import pandas as pd

force_data_path = archive_path + "/inputDirectory/postProcessing/forceCoeffs1/0/forceCoeffs.dat"

# Read the file, skipping header lines and using tab separator
data = pd.read_csv(force_data_path, sep='\t', skiprows=9, header=None)
print(f"Loaded force coefficients data with shape: {data.shape}")

What this does:

to_path: Converts Tapis URI to local file system path
pandas.read_csv: Reads force coefficient data (much cleaner than manual parsing)
skiprows=9: Skips OpenFOAM header lines
sep='\t': Uses tab separator (OpenFOAM default)

Force coefficient file format:

# Column meanings in forceCoeffs.dat
columns = {
 0: "Time",
 1: "Cm (moment coefficient)",
 2: "Cd (drag coefficient)",
 3: "Cl (lift coefficient)", 
 4: "Cl(f) (front lift)",
 5: "Cl(r) (rear lift)"
}

Plot Results¶

# Plot drag coefficient (Cd) vs time
plt.plot(data.iloc[100:, 0], data.iloc[100:, 2])
plt.xlabel('Time')
plt.ylabel('$C_d$')
plt.title('Drag Coefficient vs Time')
plt.grid(True)
plt.show()

# Plot lift coefficient (Cl) vs time 
plt.plot(data.iloc[100:, 0], data.iloc[100:, 3])
plt.xlabel('Time')
plt.ylabel('$C_l$')
plt.title('Lift Coefficient vs Time')
plt.grid(True)
plt.show()

What this does:

data.iloc[100:, 0]: Time values (column 0) starting from row 100
data.iloc[100:, 2]: Drag coefficient values (column 2)
[100:]: Skips initial transient period for cleaner plots
Creates separate plots for drag and lift coefficients

Advanced plotting options:

# Create subplots for better comparison
plt.figure(figsize=(12, 5))

plt.subplot(1, 2, 1)
plt.plot(data.iloc[100:, 0], data.iloc[100:, 2], 'b-', linewidth=2)
plt.xlabel('Time (s)')
plt.ylabel('$C_d$ (Drag Coefficient)')
plt.title('Drag Coefficient vs Time')
plt.grid(True, alpha=0.3)

plt.subplot(1, 2, 2)
plt.plot(data.iloc[100:, 0], data.iloc[100:, 3], 'r-', linewidth=2)
plt.xlabel('Time (s)')
plt.ylabel('$C_l$ (Lift Coefficient)')
plt.title('Lift Coefficient vs Time')
plt.grid(True, alpha=0.3)

plt.tight_layout()
plt.show()

# Calculate final values
final_cd = float(data.iloc[-1, 2])
final_cl = float(data.iloc[-1, 3])
print(f"Final drag coefficient: {final_cd:.6f}")
print(f"Final lift coefficient: {final_cl:.6f}")

Configuration Options¶

Environment Variable Options¶

# Complete list of OpenFOAM environment variables
openfoam_options = [
 {"key": "mesh", "value": "On"}, # Generate mesh with blockMesh
 {"key": "solver", "value": "pisoFoam"}, # Solver selection
 {"key": "decomp", "value": "On"}, # Enable parallel decomposition
 {"key": "reconstruct", "value": "On"}, # Reconstruct parallel results
 {"key": "postProcess", "value": "On"}, # Run post-processing functions
]

Queue and System Options¶

# Available queues on different systems
queue_options = {
 "stampede3": {
 "development": {"max_nodes": 2, "max_time": 120}, # 2 hours, testing
 "normal": {"max_nodes": 256, "max_time": 2880}, # 48 hours, production
 "large": {"max_nodes": 512, "max_time": 1440}, # 24 hours, large jobs
 }
}

# System-specific configurations
systems = {
 "stampede3": {"cores_per_node": 48, "memory_per_node": 192000},
 "frontera": {"cores_per_node": 56, "memory_per_node": 192000},
}

Complete Job Request Example¶

# Full-featured job request showing all options
complete_job = ds.jobs.generate(
 # Required parameters
 app_id="openfoam-stampede3",
 input_dir_uri=input_uri,
 allocation="YOUR_ALLOCATION",
 
 # Resource configuration
 max_minutes=120, # 2 hours
 node_count=2, # Multiple nodes
 cores_per_node=48, # Full node utilization
 memory_mb=192000, # 192 GB RAM
 queue="normal", # Production queue
 
 # Job metadata
 job_name="wind_flow_cfd_simulation",
 description="RANS simulation of wind flow around building using OpenFOAM",
 tags=["research", "cfd", "wind-engineering", "rans", "openfoam"],
 
 # Archive configuration
 archive_system="designsafe",
 archive_path="cfd-results/wind-study", # Results go to MyData/cfd-results/wind-study/
 
 # OpenFOAM configuration
 extra_env_vars=[
 {"key": "mesh", "value": "On"},
 {"key": "solver", "value": "simpleFoam"}, # Steady-state RANS
 {"key": "decomp", "value": "On"},
 {"key": "reconstruct", "value": "On"},
 {"key": "postProcess", "value": "On"},
 ],
 
 # Advanced options
 input_dir_param_name="Case Directory",
)