Examples

Four scripts that exercise the cfdtwin API end-to-end against PyFluent's mixing_elbow.cas.h5 case. Source for each is inlined below, or available in docs/examples/ on GitHub.

Requirements: Python 3.10+, pip install cfdtwin, and a working ANSYS Fluent installation. Each script downloads the case via PyFluent's example helper on first run; subsequent runs reuse the local cache.

quickstart.py

Smallest end-to-end pipeline: project → DOE → 20 simulations → train → predict.

"""Smallest end-to-end cfdtwin pipeline.

Downloads PyFluent's mixing_elbow case, builds a 20-sample DOE, runs the
sims, trains a surrogate, and predicts at one new design point.
"""

from pathlib import Path

import cfdtwin
from ansys.fluent.core import examples

# Get the case file (cached after first download)
case_file = examples.download_file("mixing_elbow.cas.h5", "pyfluent/mixing_elbow")

# Project lives in ./quickstart_study/ — re-run won't clobber if it exists
project_path = Path("quickstart_study")
if (project_path / "project_info.json").exists():
    project = cfdtwin.Project.open(project_path)
else:
    project = cfdtwin.Project.create(project_path, name="quickstart")

project.set_case_file(case_file)

project.set_inputs({
    "cold-inlet|momentum > velocity_magnitude": (0.2, 0.6),
    "hot-inlet|momentum > velocity_magnitude":  (0.4, 1.2),
})
project.set_outputs([
    {"name": "outlet", "category": "Surface",
     "field_variables": ["temperature"]},
])

project.generate_doe(n=20, method="lhs", seed=42)

# mixing_elbow.cas.h5 is a single-precision case file.
project.connect_fluent(precision="single")
project.run_simulations(verbose=True)   # prints "sim X/N: status" lines
result = project.train(model_name="run1")

print()
print(result.summary())

pred = project.predict("run1", {
    "cold-inlet|momentum > velocity_magnitude": 0.4,
    "hot-inlet|momentum > velocity_magnitude":  0.8,
})
print(f"\nPredicted field shape: {pred.values.shape}")
print(f"Mean predicted temperature: {pred.values.mean():.2f} K")

full_workflow.py

Same pipeline as quickstart with verbose prints between stages, a batch-prediction sweep, and notes on the resulting disk layout.

"""Full mixing_elbow workflow with verbose output and result inspection.

Same pipeline as quickstart.py — extra prints between stages so you can see
what cfdtwin is doing at each step.
"""

from pathlib import Path

import cfdtwin
from ansys.fluent.core import examples

# --- 1. Get the case file -------------------------------------------------
case_file = examples.download_file("mixing_elbow.cas.h5", "pyfluent/mixing_elbow")
print(f"Case file: {case_file}\n")

# --- 2. Project setup -----------------------------------------------------
project_path = Path("elbow_study")
if (project_path / "project_info.json").exists():
    project = cfdtwin.Project.open(project_path)
    print(f"Reopened project: {project.name}")
else:
    project = cfdtwin.Project.create(project_path, name="elbow_v1")
    print(f"Created project: {project.name}")

project.set_case_file(case_file)

# Two velocity inlets, one surface output. set_inputs is declarative — no
# Fluent connection required at this stage.
project.set_inputs({
    "cold-inlet|momentum > velocity_magnitude": (0.2, 0.6),
    "hot-inlet|momentum > velocity_magnitude":  (0.4, 1.2),
})
project.set_outputs([
    {"name": "outlet", "category": "Surface",
     "field_variables": ["temperature"]},
])
print("Inputs and outputs declared.\n")

# --- 3. DOE ---------------------------------------------------------------
n_samples = project.generate_doe(n=20, method="lhs", seed=42)
print(f"DOE: {n_samples} LHS samples\n")

# --- 4. Run sims ----------------------------------------------------------
# mixing_elbow.cas.h5 is a single-precision case file.
project.connect_fluent(precision="single")
sim_result = project.run_simulations(iterations=100, verbose=True)
print()
print(sim_result.summary())
if sim_result.failed:
    print(f"Failed sim IDs: {sim_result.failed_ids}")
print()

# --- 5. Train -------------------------------------------------------------
train_result = project.train(model_name="elbow_v1", epochs=300)
print()
print(train_result.summary())
print()

best = train_result.best_model()
print(f"Best sub-model: {best.model_name}")
print(f"  test R²:    {best.test_metrics.r2:.4f}")
print(f"  test RMSE:  {best.test_metrics.rmse:.4f}")
print(f"  POD modes:  {best.n_modes}")
print(f"  variance:   {best.variance_explained:.4f}")
print()

# --- 6. Predict at a new design point -------------------------------------
pred = project.predict(train_result.model_name, {
    "cold-inlet|momentum > velocity_magnitude": 0.4,
    "hot-inlet|momentum > velocity_magnitude":  0.8,
})
print(f"Predict result:")
print(f"  values shape:      {pred.values.shape}")
print(f"  coordinates shape: {pred.coordinates.shape if pred.coordinates is not None else 'None'}")
print(f"  mean temperature:  {pred.values.mean():.2f} K")
print(f"  range:             [{pred.values.min():.2f}, {pred.values.max():.2f}] K")

# --- 7. Batch predict (a sweep across hot-inlet velocity) -----------------
sweep = [
    {"cold-inlet|momentum > velocity_magnitude": 0.3,
     "hot-inlet|momentum > velocity_magnitude":  v}
    for v in [0.5, 0.7, 0.9, 1.1]
]
sweep_pred = project.predict(train_result.model_name, sweep)
print(f"\nSweep result: {sweep_pred.values.shape}  (4 design points)")
for inp, vals in zip(sweep_pred.inputs, sweep_pred.values):
    v = inp["hot-inlet|momentum > velocity_magnitude"]
    print(f"  hot-inlet vmag {v:.1f} -> mean outlet T = {vals.mean():.2f} K")

project.disconnect_fluent()

training_tuning.py

Trains three surrogate variants (defaults; fixed POD modes; variance cutoff with a custom learning rate) and prints a side-by-side comparison. Reuses the elbow_study/ project that full_workflow.py creates, so run that first.

"""Per-output training configuration: PCA modes, NN overrides.

Assumes you've already run quickstart.py or full_workflow.py to populate
the elbow_study project with simulation data. This script trains three
different model variants on the same data with different configurations
and compares them.
"""

from pathlib import Path

import cfdtwin

project_path = Path("elbow_study")
if not (project_path / "project_info.json").exists():
    raise SystemExit(
        "Run docs/examples/full_workflow.py first to populate elbow_study with sims."
    )
project = cfdtwin.Project.open(project_path)


# Variant 1 — defaults (let cfdtwin pick POD modes by data shape)
print("\n=== Variant 1: defaults ===")
r1 = project.train(model_name="tune_defaults")
print(r1.summary())


# Variant 2 — fixed POD mode count
print("\n=== Variant 2: fixed 15 POD modes ===")
r2 = project.train(
    model_name="tune_modes15",
    outputs={
        # The model_key is "<location>_<field>"
        "outlet_temperature": {"pod": {"modes": 15}},
    },
)
print(r2.summary())


# Variant 3 — variance-driven POD + custom NN learning rate
print("\n=== Variant 3: variance=0.99 POD + LR=5e-4 ===")
r3 = project.train(
    model_name="tune_var99",
    outputs={
        "outlet_temperature": {
            "pod": {"variance": 0.99},
            "nn":  {"learning_rate": 5e-4, "hidden_layers": [128, 64, 32]},
        },
    },
    epochs=500,
)
print(r3.summary())


# --- Compare ---
print("\n=== Comparison ===")
print(f"{'variant':<20} {'modes':<6} {'test R²':<10} {'test RMSE':<12}")
print("-" * 50)
for r in [r1, r2, r3]:
    for m in r.models:
        modes = m.n_modes if m.n_modes is not None else "-"
        print(f"{r.model_name:<20} {str(modes):<6} {m.test_metrics.r2:<10.4f} {m.test_metrics.rmse:<12.4g}")

discovering_bcs.py

Launches Fluent and lists every boundary condition, settable parameter path, surface, and report definition exposed by the case. Useful when starting from an unfamiliar case file. Also demonstrates the Project.list_available_inputs() shortcut.

"""Connect to Fluent and list everything cfdtwin can address as inputs/outputs.

Run this once for an unfamiliar case file — copy the bc names, parameter
paths, surface names, etc. into your set_inputs / set_outputs calls.

The last section shows the higher-level ``Project.list_available_inputs()``
helper, which gives you the same data already grouped by category and parsed
into a shape you can feed straight into ``set_inputs``.
"""

import tempfile

import ansys.fluent.core as pyfluent
from ansys.fluent.core import examples

import cfdtwin

CASE_FILE = examples.download_file("mixing_elbow.cas.h5", "pyfluent/mixing_elbow")

print(f"Loading {CASE_FILE}...\n")
solver = pyfluent.launch_fluent(
    precision="single", processor_count=4, dimension=3, mode="solver",
)
solver.settings.file.read_case(file_name=CASE_FILE)

# --- Boundary conditions --------------------------------------------------
print("=== Boundary conditions ===")
bc_settings = solver.settings.setup.boundary_conditions
for bc_type in bc_settings.get_state():
    bc_group = getattr(bc_settings, bc_type)
    for bc_name in bc_group.get_state():
        print(f"  {bc_name:<24}  type={bc_type}")


def walk_settable(obj, prefix="", depth=4):
    """Recursively find leaf settings that have a writable .value, yielding
    (dotted-path, type-name) pairs. Used to show what set_inputs's rich-dict
    `parameter_path` can target on a given BC.

    Skips PyFluent nodes that raise InactiveObjectError (settings only
    relevant under another flag — e.g. gauge_pressure is inactive when the
    BC isn't pressure-driven).
    """
    if depth <= 0:
        return
    try:
        names = obj.child_names
    except Exception:
        return
    for name in names:
        if name in ("child_names", "command_names"):
            continue
        path = f"{prefix}.{name}" if prefix else name
        try:
            child = getattr(obj, name)
            if hasattr(child, "value"):
                yield path, type(child).__name__
                continue
            yield from walk_settable(child, prefix=path, depth=depth - 1)
        except Exception:
            # Inactive node, or settings tree disagrees with what's exposed
            # at this point in the case. Either way, skip silently.
            continue


# --- Settable parameter paths on each velocity-inlet ---------------------
print("\n=== Velocity-inlet settable parameters ===")
vel_inlets = solver.settings.setup.boundary_conditions.velocity_inlet
for bc_name in vel_inlets.get_state():
    print(f"  {bc_name}:")
    inlet = vel_inlets[bc_name]
    for path, type_name in walk_settable(inlet):
        print(f"    parameter_path={path!r:<48} ({type_name})")

# --- User-defined surfaces -----------------------------------------------
# results.surfaces is grouped by surface type (point_surface, zone_surface,
# plane_surface, ...) — iterate one level deeper to get the actual names.
print("\n=== Surfaces ===")
try:
    surfaces = solver.settings.results.surfaces.get_state() or {}
    any_found = False
    for surf_type, items in surfaces.items():
        if not items:
            continue
        for surf_name in items:
            print(f"  {surf_name:<24}  type={surf_type}")
            any_found = True
    if not any_found:
        print("  (no user-defined surfaces; BC names like 'outlet' still work as outputs)")
except Exception as e:
    print(f"  (unable to list: {e})")

# --- User-defined report definitions -------------------------------------
# Same shape as surfaces: top-level keys are report types, each holding a
# dict of user-created reports.
print("\n=== Report definitions ===")
try:
    reports = solver.settings.solution.report_definitions.get_state() or {}
    any_found = False
    for report_type, items in reports.items():
        if not items:
            continue
        for report_name in items:
            print(f"  {report_name:<24}  type={report_type}")
            any_found = True
    if not any_found:
        print("  (none defined in this case)")
except Exception as e:
    print(f"  (unable to list: {e})")

solver.exit()

# --- Same data via the cfdtwin API ----------------------------------------
# Project.list_available_inputs() wraps the raw PyFluent calls above. It
# returns one list with BCs and Fluent input parameters (named expressions
# flagged input_parameter=True) already merged, each entry tagged with a
# `category`. Feed entries straight back into set_inputs by attaching a range.
print("\n=== cfdtwin.Project.list_available_inputs() ===")
with tempfile.TemporaryDirectory() as tmp:
    project = cfdtwin.Project.create(tmp, name="discovery_demo")
    project.set_case_file(CASE_FILE)
    project.connect_fluent(precision="single")
    try:
        for item in project.list_available_inputs():
            cat = item['category']
            if cat == 'Input Parameter':
                print(f"  [IP] {item['name']:<24} unit={item.get('unit','')} "
                      f"current={item.get('current_value')}")
            else:
                print(f"  [BC] {item['name']:<24} type={item['type']}")
    finally:
        project.disconnect_fluent()