📁 Case manipulation

class foamlib.FoamCaseBase(path: PathLike[str] | str = PosixPath('.'))

Bases: Sequence[FoamCaseBase.TimeDirectory], PathLike[str]

Base class for OpenFOAM cases.

Provides methods for accessing files and time directories in the case, but does not provide methods for running the case or any commands. Users are encouraged to use FoamCase or AsyncFoamCase instead of this class.

Access the time directories of the case as a sequence, e.g. case[0] or case[-1]. These will return FoamCaseBase.TimeDirectory objects.

Parameters:: path – The path to the case directory. Defaults to the current working directory.

property name: str: The name of the case.

property control_dict: FoamFile: The controlDict file.

property block_mesh_dict: FoamFile: The blockMeshDict file.

property decompose_par_dict: FoamFile: The decomposeParDict file.

property fv_schemes: FoamFile: The fvSchemes file.

property fv_solution: FoamFile: The fvSolution file.

property transport_properties: FoamFile: The transportProperties file.

property turbulence_properties: FoamFile: The turbulenceProperties file.

property application: str: The application name.

file(path: PathLike[str] | str) → FoamFile: Return a FoamFile object for the given path in the case.

__getitem__(index: int | float | str) → TimeDirectory
__getitem__(index: slice) → Sequence[TimeDirectory]: Return the time directory at the given index (int), indices (slice), name (str), or time (float).

__iter__() → Iterator[TimeDirectory]: Return an iterator over the time directories in the case.

__contains__(value: object) → bool: Return True if the given time directory, name, or time exists in the case.

__len__() → int: Return the number of time directories in the case.

__delitem__(key: int | float | str, /) → None: Delete the time directory at the given index (int), name (str), or time (float).

class TimeDirectory(path: PathLike[str] | str)

Bases: Set[FoamFieldFile], PathLike[str]

A time directory in an OpenFOAM case.

Use to access field files in the directory (e.g. time["U"]). These will be returned as FoamFieldFile objects.

It also behaves as a set of FoamFieldFile objects (e.g. it can be iterated over with for field in time: ...).

property TimeDirectory.name: str: The name of this time directory (the time as a string).

property TimeDirectory.time: float: The time that corresponds to this directory, as a float.

TimeDirectory.__getitem__(key: str, /) → FoamFieldFile: Return the field file with the given name in this time directory.

TimeDirectory.__iter__() → Iterator[FoamFieldFile]: Return an iterator over the field files in this time directory.

TimeDirectory.__contains__(x: object) → bool: Return True if the given field file or name exists in this time directory.

TimeDirectory.__len__() → int: Return the number of field files in this time directory.

TimeDirectory.__delitem__(name: str, /) → None: Delete the field file with the given name in this time directory.

class foamlib.FoamCase(path: PathLike[str] | str = PosixPath('.'))

Bases: FoamCaseRunBase

An OpenFOAM case with synchronous execution capabilities.

Extends FoamCaseBase to provide methods for running and cleaning cases, as well as accessing files.

Access the time directories of the case as a sequence, e.g. case[0] or case[-1]. These will return FoamCase.TimeDirectory objects.

Parameters:: path – The path to the case directory. Defaults to the current working directory.

Example usage:

from foamlib import FoamCase

case = FoamCase("path/to/case") # Load an OpenFOAM case
case[0]["U"].internal_field = [0, 0, 0] # Set the initial velocity field to zero
case.run() # Run the case
for time in case: # Iterate over the time directories
    print(time.time) # Print the time
    print(time["U"].internal_field) # Print the velocity field

class TimeDirectory(path: PathLike[str] | str)

Bases: TimeDirectory

cell_centers() → FoamFieldFile

Write and return the cell centers.

Currently only works for reconstructed cases (decomposed cases will need to be reconstructed first).

block_mesh(*, check: bool = True, log: bool | str | PathLike[str] = True) → None: Run blockMesh on this case.

clean(*, check: bool = False) → None

Clean this case.

If a clean or Allclean script is present in the case directory, it will be invoked. Otherwise, the case directory will be cleaned using these rules:

All time directories except 0 will be deleted.
The 0 time directory will be deleted if 0.orig exists.
processor* directories will be deleted if a system/decomposeParDict file is present.
constant/polyMesh will be deleted if a system/blockMeshDict file is present.
All log.* files will be deleted.

If this behavior is not appropriate for a case, it is recommended to write a custom clean script.

Parameters:: check – If True, raise a CalledProcessError if the clean script returns a non-zero exit code.

clone(dst: PathLike[str] | str | None = None) → Self

Clone this case (make a clean copy).

This is equivalent to running self.copy().clean(), but it can be more efficient in cases that do not contain custom clean scripts.

If used as a context manager (i.e., within a with block) the cloned copy will be deleted automatically when exiting the block.

Parameters:: dst – The destination path. If None, clone to a new directory in $FOAM_RUN/foamlib.
Returns:: The clone of the case.

Example usage:

import os
from pathlib import Path
from foamlib import FoamCase

pitz_tutorial = FoamCase(Path(os.environ["FOAM_TUTORIALS"]) / "incompressible/simpleFoam/pitzDaily")

my_pitz = pitz_tutorial.clone("myPitz")

copy(dst: PathLike[str] | str | None = None) → Self

Make a copy of this case.

If used as a context manager (i.e., within a with block) the copy will be deleted automatically when exiting the block.

Parameters:: dst – The destination path. If None, copy to a new directory in $FOAM_RUN/foamlib.
Returns:: The copy of the case.

Example usage:

import os
from pathlib import Path
from foamlib import FoamCase

pitz_tutorial = FoamCase(Path(os.environ["FOAM_TUTORIALS"]) / "incompressible/simpleFoam/pitzDaily")

my_pitz = pitz_tutorial.copy("myPitz")

decompose_par(*, check: bool = True, log: bool | str | PathLike[str] = True) → None: Decompose this case for parallel running.

reconstruct_par(*, check: bool = True, log: bool | str | PathLike[str] = True) → None: Reconstruct this case after parallel running.

restore_0_dir() → None: Restore the 0 directory from the 0.orig directory.

Run this case, or a specified command in the context of this case.

If cmd is given, this method will run the given command in the context of the case.

If cmd is None, a series of heuristic rules will be used to run the case. This works as follows:

If a run, Allrun or Allrun-parallel script is present in the case directory, it will be invoked. If both run and Allrun are present, Allrun will be used. If both Allrun and Allrun-parallel are present and :param:`parallel` is None, an error will be raised.
If no run script is present but an Allrun.pre script exists, it will be invoked.
Otherwise, if a system/blockMeshDict file is present, the method will call block_mesh().
Then, if a 0.orig directory is present, it will call restore_0_dir().
Then, if the case is to be run in parallel (see the :param:`parallel` option) and no processor* directories exist but a system/decomposeParDict file is present, it will call decompose_par().
Then, it will run the case using the application specified in the controlDict file.

If this behavior is not appropriate for a case, it is recommended to write a custom run, Allrun, Allrun-parallel or Allrun.pre script.

Parameters:

cmd – The command to run. If None, run the case. If a sequence, the first element is the command and the rest are arguments. If a string, cmd is executed in a shell.
parallel – If True, run in parallel using MPI. If None, autodetect whether to run in parallel.
cpus – The number of CPUs to use. If None, autodetect from to the case.
check – If True, raise a CalledProcessError if any command returns a non-zero exit code.
log – If True, log the command output to log.* files in the case directory.

class foamlib.AsyncFoamCase(path: PathLike[str] | str = PosixPath('.'))

Bases: FoamCaseRunBase

An OpenFOAM case with asynchronous execution capabilities.

Extends FoamCaseBase with methods for running and cleaning cases asynchronously. This allows for non-blocking execution and parallel execution of multiple cases.

Access the time directories of the case as a sequence, e.g. case[0] or case[-1]. These will return AsyncFoamCase.TimeDirectory objects.

Parameters:: path – The path to the case directory. Defaults to the current working directory.

Example usage:

from foamlib import AsyncFoamCase

case = AsyncFoamCase("path/to/case") # Load an OpenFOAM case
case[0]["U"].internal_field = [0, 0, 0] # Set the initial velocity field to zero
await case.run() # Run the case
for time in case: # Iterate over the time directories
    print(time.time) # Print the time
    print(time["U"].internal_field) # Print the velocity field

class TimeDirectory(path: PathLike[str] | str)

Bases: TimeDirectory

async cell_centers() → FoamFieldFile

Write and return the cell centers.

Currently only works for reconstructed cases (decomposed cases will need to be reconstructed first).

async block_mesh(*, check: bool = True, log: bool | str | PathLike[str] = True) → None: Run blockMesh on this case.

async clean(*, check: bool = False) → None

Clean this case.

If a clean or Allclean script is present in the case directory, it will be invoked. Otherwise, the case directory will be cleaned using these rules:

All time directories except 0 will be deleted.
The 0 time directory will be deleted if 0.orig exists.
processor* directories will be deleted if a system/decomposeParDict file is present.
constant/polyMesh will be deleted if a system/blockMeshDict file is present.
All log.* files will be deleted.

If this behavior is not appropriate for a case, it is recommended to write a custom clean script.

Parameters:: check – If True, raise a CalledProcessError if the clean script returns a non-zero exit code.

clone(dst: PathLike[str] | str | None = None) → AsyncGenerator[Self]

Clone this case (make a clean copy).

This is equivalent to running (await self.copy()).clean(), but it can be more efficient in cases that do not contain custom clean scripts.

If used as an asynchronous context manager (i.e., within an async with block) the cloned copy will be deleted automatically when exiting the block.

Parameters:: dst – The destination path. If None, clone to a new directory in $FOAM_RUN/foamlib.
Returns:: The clone of the case.

Example usage:

import os
from pathlib import Path
from foamlib import AsyncFoamCase

pitz_tutorial = AsyncFoamCase(Path(os.environ["FOAM_TUTORIALS"]) / "incompressible/simpleFoam/pitzDaily")

my_pitz = await pitz_tutorial.clone("myPitz")

copy(dst: PathLike[str] | str | None = None) → AsyncGenerator[Self]

Make a copy of this case.

If used as an asynchronous context manager (i.e., within an async with block) the copy will be deleted automatically when exiting the block.

Parameters:: dst – The destination path. If None, copy to a new directory in $FOAM_RUN/foamlib.
Returns:: The copy of the case.

Example usage:

import os
from pathlib import Path
from foamlib import AsyncFoamCase

pitz_tutorial = AsyncFoamCase(Path(os.environ["FOAM_TUTORIALS"]) / "incompressible/simpleFoam/pitzDaily")

my_pitz = await pitz_tutorial.copy("myPitz")

async decompose_par(*, check: bool = True, log: bool | str | PathLike[str] = True) → None: Decompose this case for parallel running.

static map(coro: Callable[[_X], Awaitable[_Y]], iterable: Iterable[_X], /) → Iterable[_Y]

Run an async function on each element of an iterable concurrently.

Note that if coro runs AsyncFoamCase`s, these will be executed in parallel if possible (this parallelism is automatically limited by the :attr:`max_cpus attribute in order to avoid oversubscription of available CPUs).

Parameters:

coro – An async function to run on each element.
iterable – An iterable of arguments to pass to the function.

Returns:

An iterable of results from the function.

Example usage:

import os
from pathlib import Path
from foamlib import AsyncSlurmFoamCase
from scipy.optimize import differential_evolution

# Set up base case for optimization
base = AsyncSlurmFoamCase(Path(os.environ["FOAM_TUTORIALS"]) / "incompressible/simpleFoam/pitzDaily")

async def objective_function(x):
    async with base.clone() as case:
        # Set inlet velocity based on optimization parameters
        case[0]["U"].boundary_field["inlet"].value = [x[0], 0, 0]

        # Run with fallback to local execution if Slurm unavailable
        await case.run(fallback=True)

        # Return objective (minimize velocity magnitude at outlet)
        return abs(case[-1]["U"].internal_field[0][0])

# Run optimization with parallel jobs
result = differential_evolution(
    objective_function,
    bounds=[(-1, 1)],
    workers=AsyncSlurmFoamCase.map,
    polish=False
)
print(f"Optimal inlet velocity: {result.x[0]}")

max_cpus = 2

Maximum number of CPUs to use for running instances of AsyncFoamCase concurrently.

Defaults to the number of CPUs on the system.

async reconstruct_par(*, check: bool = True, log: bool | str | PathLike[str] = True) → None: Reconstruct this case after parallel running.

async restore_0_dir() → None: Restore the 0 directory from the 0.orig directory.

Run this case, or a specified command in the context of this case.

If cmd is given, this method will run the given command in the context of the case.

If cmd is None, a series of heuristic rules will be used to run the case. This works as follows:

If a run, Allrun or Allrun-parallel script is present in the case directory, it will be invoked. If both run and Allrun are present, Allrun will be used. If both Allrun and Allrun-parallel are present and parallel is None, an error will be raised.
If no run script is present but an Allrun.pre script exists, it will be invoked.
Otherwise, if a system/blockMeshDict file is present, the method will call block_mesh().
Then, if a 0.orig directory is present, it will call restore_0_dir().
Then, if the case is to be run in parallel (see the parallel option) and no processor* directories exist but a system/decomposeParDict file is present, it will call decompose_par().
Then, it will run the case using the application specified in the controlDict file.

If this behavior is not appropriate for a case, it is recommended to write a custom run, Allrun, Allrun-parallel or Allrun.pre script.

Parameters:

cmd – The command to run. If None, run the case. If a sequence, the first element is the command and the rest are arguments. If a string, cmd is executed in a shell.
parallel – If True, run in parallel using MPI. If None, autodetect whether to run in parallel.
cpus – The number of CPUs to use. If None, autodetect from to the case.
check – If True, raise a CalledProcessError if any command returns a non-zero exit code.
log – If True, log the command output to log.* files in the case directory.

async static run_all(cases: Iterable[AsyncFoamCase | Awaitable[object]], /) → None

Run multiple cases concurrently.

Multiple cases will be run in parallel if possible (this parallelism is automatically limited by the max_cpus attribute in order to avoid oversubscription of available CPUs).

Parameters:: cases – Instances of AsyncFoamCase (run() will be called) or arbitrary awaitables (will be awaited as-is).

Example usage:

from foamlib import AsyncFoamCase

case1 = AsyncFoamCase("path/to/case1")
case2 = AsyncFoamCase("path/to/case2")

await AsyncFoamCase.run_all([case1, case2])

class foamlib.AsyncSlurmFoamCase(path: PathLike[str] | str = PosixPath('.'))

Bases: AsyncFoamCase

An asynchronous OpenFOAM case that launches jobs on a Slurm cluster.

AsyncSlurmFoamCase is a subclass of AsyncFoamCase. It provides the same interface, as the latter, except that it will launch jobs on a Slurm cluster (using salloc and srun) on the user’s behalf when running a case or command.

Parameters:: path – The path to the case directory. Defaults to the current working directory.

Run this case, or a specified command in the context of this case.

Parameters:

cmd – The command to run. If None, run the case. If a sequence, the first element is the command and the rest are arguments. If a string, cmd is executed in a shell.
parallel – If True, run in parallel using MPI. If None, autodetect whether to run in parallel.
cpus – The number of CPUs to use. If None, autodetect according to the case. If 0, run locally.
check – If True, raise a CalledProcessError if any command returns a non-zero exit code.
log – If True, log the command output to a file.
fallback – If True, fall back to running the command locally if Slurm is not available.

Exceptions

class foamlib.CalledProcessError(returncode, cmd, output=None, stderr=None)