Skip to content

Boundary conditions

jax_lab.core.boundary_conditions.BoundaryCondition

Bases: object

Base class for boundary conditions in a LBM simulation.

This class provides a general structure for implementing boundary conditions. It includes methods for preparing the boundary attributes and for applying the boundary condition. Specific boundary conditions should be implemented as subclasses of this class, with the apply method overridden as necessary.

Attributes

lattice (Lattice): The lattice used in the simulation.

nx (int): The number of nodes in the x direction.

ny (int): The number of nodes in the y direction.

nz (int): The number of nodes in the z direction.

dim (int): The number of dimensions in the simulation (2 or 3).

precision_policy (PrecisionPolicy): The precision policy used in the simulation.

indices (array-like): The indices of the boundary nodes.

name (str or None): The name of the boundary condition. This should be set in subclasses.

is_solid (bool): Whether the boundary condition is for a solid boundary. This should be set in subclasses.

is_dynamic (bool): Whether the boundary condition is dynamic (changes over time). This should be set in subclasses.

needs_extra_configuration (bool): Whether the boundary condition requires extra configuration. This should be set in subclasses.

implementation_step (str): The lattice Boltzmann algorithm step at which the boundary condition is applied.

Source code in jax_lab/core/boundary_conditions.py
class BoundaryCondition(object):
    """
    Base class for boundary conditions in a LBM simulation.

    This class provides a general structure for implementing boundary conditions. It includes methods for preparing the
    boundary attributes and for applying the boundary condition. Specific boundary conditions should be implemented as
    subclasses of this class, with the `apply` method overridden as necessary.

    Attributes
    ----------
    lattice (Lattice): The lattice used in the simulation.

    nx (int): The number of nodes in the x direction.

    ny (int): The number of nodes in the y direction.

    nz (int): The number of nodes in the z direction.

    dim (int): The number of dimensions in the simulation (2 or 3).

    precision_policy (PrecisionPolicy): The precision policy used in the simulation.

    indices (array-like): The indices of the boundary nodes.

    name (str or None): The name of the boundary condition. This should be set in subclasses.

    is_solid (bool): Whether the boundary condition is for a solid boundary. This should be set in subclasses.

    is_dynamic (bool): Whether the boundary condition is dynamic (changes over time). This should be set in subclasses.

    needs_extra_configuration (bool): Whether the boundary condition requires extra configuration. This should be set in subclasses.

    implementation_step (str): The lattice Boltzmann algorithm step at which the boundary condition is applied.
    """

    def __init__(self, indices, grid_info, precision_policy):
        self.lattice = grid_info["lattice"]
        self.nx = grid_info["nx"]
        self.ny = grid_info["ny"]
        self.nz = grid_info["nz"]
        self.dim = grid_info["dim"]
        self.precision_policy = precision_policy
        self.indices = indices
        self.name = None
        self.is_solid = False
        self.is_dynamic = False
        self.needs_extra_configuration = False
        self.implementation_step = "PostStreaming"

    def create_local_mask_and_normal_arrays(self, grid_mask):
        """
        Creates local mask and normal arrays for the boundary condition.

        Parameters
        ----------
        grid_mask (array-like): The grid mask for the lattice.

        Returns
        -------
        None

        Notes
        -----
        This method creates local mask and normal arrays for the boundary condition based on the grid mask.
        If the boundary condition requires extra configuration, the `configure` method is called.
        """

        if self.needs_extra_configuration:
            boundaryMask = self.get_boundary_mask(grid_mask)
            self.configure(boundaryMask)
            self.needs_extra_configuration = False

        boundaryMask = self.get_boundary_mask(grid_mask)
        self.normals = self.get_normals(boundaryMask)
        self.imissing, self.iknown = self.get_missing_indices(boundaryMask)
        self.imissing_mask, self.iknown_mask, self.imiddle_mask = self.get_missing_mask(boundaryMask)

        return

    def get_boundary_mask(self, grid_mask):
        """
        Add jax.device_count() to the self.indices in x-direction, and 1 to the self.indices other directions
        This is to make sure the boundary condition is applied to the correct nodes as grid_mask is
        expanded by (jax.device_count(), 1, 1)

        Parameters
        ----------
        grid_mask (array-like): The grid mask for the lattice.

        Returns
        -------
        boundaryMask : (array-like)
        """
        shifted_indices = np.array(self.indices)
        shifted_indices[0] += device_count()
        shifted_indices[1:] += 1
        # Convert back to tuple
        shifted_indices = tuple(shifted_indices)
        boundaryMask = np.array(grid_mask[shifted_indices])

        return boundaryMask

    def configure(self, boundary_mask):
        """
        Configures the boundary condition.

        Parameters
        ----------
        boundary_mask (array-like): The grid mask for the boundary voxels.

        Returns
        -------
        None

        Notes
        -----
        This method should be overridden in subclasses if the boundary condition requires extra configuration.
        """
        return

    @partial(jit, static_argnums=(0, 3), inline=True)
    def prepare_populations(self, fout, fin, implementation_step):
        """
        Prepares the distribution functions for the boundary condition.

        Parameters
        ----------
        fout (jax.numpy.ndarray): The incoming distribution functions.

        fin (jax.numpy.ndarray): The outgoing distribution functions.

        implementation_step (str): The step in the lattice Boltzmann method algorithm at which the preparation is
            applied.

        Returns
        -------
        (jax.numpy.ndarray): The prepared distribution functions.

        Notes
        -----
        This method should be overridden in subclasses if the boundary condition requires preparation of the distribution functions during post-collision or post-streaming.
        See ExtrapolationBoundaryCondition for an example.
        """
        return fout

    def get_normals(self, boundary_mask):
        """
        Calculates the normal vectors at the boundary nodes.

        Parameters
        ----------
        boundary_mask (array-like): The boundary mask for the lattice.

        Returns
        -------
        (array-like): The normal vectors at the boundary nodes.

        Notes
        -----
        This method calculates the normal vectors by dotting the boundary mask with the main lattice directions.
        """
        main_c = self.lattice.c.T[self.lattice.main_indices]
        m = boundary_mask[..., self.lattice.main_indices]
        normals = -np.dot(m, main_c)
        return normals

    def get_missing_indices(self, boundary_mask):
        """
        Returns two int8 arrays the same shape as boundary_mask. The non-zero entries of these arrays indicate missing
        directions that require BCs (imissing) as well as their corresponding opposite directions (iknown).

        Parameters
        ----------
        boundary_mask (array-like): The boundary mask for the lattice.

        Returns
        -------
        (tuple of array-like): The missing and known indices for the boundary condition.

        Notes
        -----
        This method calculates the missing and known indices based on the boundary mask. The missing indices are the
        non-zero entries of the boundary mask, and the known indices are their corresponding opposite directions.
        """

        # Find imissing, iknown 1-to-1 corresponding indices
        # Note: the "zero" index is used as default value here and won't affect BC computations
        nbd = len(self.indices[0])
        imissing = np.vstack([np.arange(self.lattice.q, dtype="uint8")] * nbd)
        iknown = np.vstack([self.lattice.opp_indices] * nbd)
        imissing[~boundary_mask] = 0
        iknown[~boundary_mask] = 0
        return imissing, iknown

    def get_missing_mask(self, boundary_mask):
        """
        Returns three boolean arrays the same shape as boundary_mask.
        Note: these boundary masks are useful for reduction (eg. summation) operators of selected q-directions.

        Parameters
        ----------
        boundary_mask (array-like): The boundary mask for the lattice.

        Returns
        -------
        (tuple of array-like): The missing, known, and middle masks for the boundary condition.

        Notes
        -----
        This method calculates the missing, known, and middle masks based on the boundary mask. The missing mask
        is the boundary mask, the known mask is the opposite directions of the missing mask, and the middle mask
        is the directions that are neither missing nor known.
        """
        # Find masks for imissing, iknown and imiddle
        imissingMask = boundary_mask
        iknownMask = imissingMask[:, self.lattice.opp_indices]
        imiddleMask = ~(imissingMask | iknownMask)
        return imissingMask, iknownMask, imiddleMask

    @partial(jit, static_argnums=(0,))
    def apply(self, fout, fin):
        """
        Applies the boundary condition.

        Parameters
        ----------
        fout (jax.Array): Output distribution functions.

        fin (jax.Array): Input distribution functions.

        Returns
        -------
        None

        Notes
        -----
        This method should be overridden in subclasses to implement the specific boundary condition. The method should
        modify the output distribution functions in place to apply the boundary condition.
        """
        pass

    @partial(jit, static_argnums=(0,))
    def equilibrium(self, rho, u):
        """
        Compute equilibrium distribution function.

        Parameters
        ----------
        rho (jax.numpy.ndarray): The density at each node in the lattice.

        u (jax.numpy.ndarray): The velocity at each node in the lattice.

        Returns
        -------
        (jax.numpy.ndarray): The equilibrium distribution function at each node in the lattice.

        Notes
        -----
        This method computes the equilibrium distribution function based on the density and velocity. The computation is
        performed in the compute precision specified by the precision policy. The result is not cast to the output precision as
        this is function is used inside other functions that require the compute precision.
        """
        rho, u = self.precision_policy.cast_to_compute((rho, u))
        c = jnp.array(self.lattice.c, dtype=self.precision_policy.compute_dtype)
        cu = 3.0 * jnp.dot(u, c)
        usqr = 1.5 * jnp.sum(u**2, axis=-1, keepdims=True)
        feq = rho * self.lattice.w * (1.0 + 1.0 * cu + 0.5 * cu**2 - usqr)

        return feq

    @partial(jit, static_argnums=(0,))
    def momentum_flux(self, fneq):
        """
        Compute the momentum flux.

        Parameters
        ----------
        fneq (jax.numpy.ndarray): The non-equilibrium distribution function at each node in the lattice.

        Returns
        -------
        (jax.numpy.ndarray): The momentum flux at each node in the lattice.

        Notes
        -----
        This method computes the momentum flux by dotting the non-equilibrium distribution function with the lattice
        direction vectors.
        """
        return jnp.dot(fneq, self.lattice.cc)

    @partial(jit, static_argnums=(0,))
    def momentum_exchange_force(self, f_poststreaming, f_postcollision):
        """
        Using the momentum exchange method to compute the boundary force vector exerted on the solid geometry
        based on [1] as described in [3]. Ref [2] shows how [1] is applicable to curved geometries only by using a
        bounce-back method (e.g. Bouzidi) that accounts for curved boundaries.
        NOTE: this function should be called after BC's are imposed.
        [1] A.J.C. Ladd, Numerical simulations of particular suspensions via a discretized Boltzmann equation.
            Part 2 (numerical results), J. Fluid Mech. 271 (1994) 311-339.
        [2] R. Mei, D. Yu, W. Shyy, L.-S. Luo, Force evaluation in the lattice Boltzmann method involving
            curved geometry, Phys. Rev. E 65 (2002) 041203.
        [3] Caiazzo, A., & Junk, M. (2008). Boundary forces in lattice Boltzmann: Analysis of momentum exchange
            algorithm. Computers & Mathematics with Applications, 55(7), 1415-1423.

        Parameters
        ----------
        f_poststreaming (jax.numpy.ndarray): The post-streaming distribution function at each node in the lattice.

        f_postcollision (jax.numpy.ndarray): The post-collision distribution function at each node in the lattice.

        Returns
        -------
        (jax.numpy.ndarray): The force exerted on the solid geometry at each boundary node.

        Notes
        -----
        This method computes the force exerted on the solid geometry at each boundary node using the momentum exchange method.
        The force is computed based on the post-streaming and post-collision distribution functions. This method
        should be called after the boundary conditions are imposed.
        """
        c = jnp.array(self.lattice.c, dtype=self.precision_policy.compute_dtype)
        nbd = len(self.indices[0])
        bindex = np.arange(nbd)[:, None]
        phi = f_postcollision[self.indices][bindex, self.iknown] + f_poststreaming[self.indices][bindex, self.imissing]
        force = jnp.sum(c[:, self.iknown] * phi, axis=-1).T
        return force

apply

apply(fout, fin)

Applies the boundary condition.

Parameters

fout (jax.Array): Output distribution functions.

fin (jax.Array): Input distribution functions.

Returns

None

Notes

This method should be overridden in subclasses to implement the specific boundary condition. The method should modify the output distribution functions in place to apply the boundary condition.

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0,))
def apply(self, fout, fin):
    """
    Applies the boundary condition.

    Parameters
    ----------
    fout (jax.Array): Output distribution functions.

    fin (jax.Array): Input distribution functions.

    Returns
    -------
    None

    Notes
    -----
    This method should be overridden in subclasses to implement the specific boundary condition. The method should
    modify the output distribution functions in place to apply the boundary condition.
    """
    pass

configure

configure(boundary_mask)

Configures the boundary condition.

Parameters

boundary_mask (array-like): The grid mask for the boundary voxels.

Returns

None

Notes

This method should be overridden in subclasses if the boundary condition requires extra configuration.

Source code in jax_lab/core/boundary_conditions.py
def configure(self, boundary_mask):
    """
    Configures the boundary condition.

    Parameters
    ----------
    boundary_mask (array-like): The grid mask for the boundary voxels.

    Returns
    -------
    None

    Notes
    -----
    This method should be overridden in subclasses if the boundary condition requires extra configuration.
    """
    return

create_local_mask_and_normal_arrays

create_local_mask_and_normal_arrays(grid_mask)

Creates local mask and normal arrays for the boundary condition.

Parameters

grid_mask (array-like): The grid mask for the lattice.

Returns

None

Notes

This method creates local mask and normal arrays for the boundary condition based on the grid mask. If the boundary condition requires extra configuration, the configure method is called.

Source code in jax_lab/core/boundary_conditions.py
def create_local_mask_and_normal_arrays(self, grid_mask):
    """
    Creates local mask and normal arrays for the boundary condition.

    Parameters
    ----------
    grid_mask (array-like): The grid mask for the lattice.

    Returns
    -------
    None

    Notes
    -----
    This method creates local mask and normal arrays for the boundary condition based on the grid mask.
    If the boundary condition requires extra configuration, the `configure` method is called.
    """

    if self.needs_extra_configuration:
        boundaryMask = self.get_boundary_mask(grid_mask)
        self.configure(boundaryMask)
        self.needs_extra_configuration = False

    boundaryMask = self.get_boundary_mask(grid_mask)
    self.normals = self.get_normals(boundaryMask)
    self.imissing, self.iknown = self.get_missing_indices(boundaryMask)
    self.imissing_mask, self.iknown_mask, self.imiddle_mask = self.get_missing_mask(boundaryMask)

    return

equilibrium

equilibrium(rho, u)

Compute equilibrium distribution function.

Parameters

rho (jax.numpy.ndarray): The density at each node in the lattice.

u (jax.numpy.ndarray): The velocity at each node in the lattice.

Returns

(jax.numpy.ndarray): The equilibrium distribution function at each node in the lattice.

Notes

This method computes the equilibrium distribution function based on the density and velocity. The computation is performed in the compute precision specified by the precision policy. The result is not cast to the output precision as this is function is used inside other functions that require the compute precision.

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0,))
def equilibrium(self, rho, u):
    """
    Compute equilibrium distribution function.

    Parameters
    ----------
    rho (jax.numpy.ndarray): The density at each node in the lattice.

    u (jax.numpy.ndarray): The velocity at each node in the lattice.

    Returns
    -------
    (jax.numpy.ndarray): The equilibrium distribution function at each node in the lattice.

    Notes
    -----
    This method computes the equilibrium distribution function based on the density and velocity. The computation is
    performed in the compute precision specified by the precision policy. The result is not cast to the output precision as
    this is function is used inside other functions that require the compute precision.
    """
    rho, u = self.precision_policy.cast_to_compute((rho, u))
    c = jnp.array(self.lattice.c, dtype=self.precision_policy.compute_dtype)
    cu = 3.0 * jnp.dot(u, c)
    usqr = 1.5 * jnp.sum(u**2, axis=-1, keepdims=True)
    feq = rho * self.lattice.w * (1.0 + 1.0 * cu + 0.5 * cu**2 - usqr)

    return feq

get_boundary_mask

get_boundary_mask(grid_mask)

Add jax.device_count() to the self.indices in x-direction, and 1 to the self.indices other directions This is to make sure the boundary condition is applied to the correct nodes as grid_mask is expanded by (jax.device_count(), 1, 1)

Parameters

grid_mask (array-like): The grid mask for the lattice.

Returns

boundaryMask : (array-like)

Source code in jax_lab/core/boundary_conditions.py
def get_boundary_mask(self, grid_mask):
    """
    Add jax.device_count() to the self.indices in x-direction, and 1 to the self.indices other directions
    This is to make sure the boundary condition is applied to the correct nodes as grid_mask is
    expanded by (jax.device_count(), 1, 1)

    Parameters
    ----------
    grid_mask (array-like): The grid mask for the lattice.

    Returns
    -------
    boundaryMask : (array-like)
    """
    shifted_indices = np.array(self.indices)
    shifted_indices[0] += device_count()
    shifted_indices[1:] += 1
    # Convert back to tuple
    shifted_indices = tuple(shifted_indices)
    boundaryMask = np.array(grid_mask[shifted_indices])

    return boundaryMask

get_missing_indices

get_missing_indices(boundary_mask)

Returns two int8 arrays the same shape as boundary_mask. The non-zero entries of these arrays indicate missing directions that require BCs (imissing) as well as their corresponding opposite directions (iknown).

Parameters

boundary_mask (array-like): The boundary mask for the lattice.

Returns

(tuple of array-like): The missing and known indices for the boundary condition.

Notes

This method calculates the missing and known indices based on the boundary mask. The missing indices are the non-zero entries of the boundary mask, and the known indices are their corresponding opposite directions.

Source code in jax_lab/core/boundary_conditions.py
def get_missing_indices(self, boundary_mask):
    """
    Returns two int8 arrays the same shape as boundary_mask. The non-zero entries of these arrays indicate missing
    directions that require BCs (imissing) as well as their corresponding opposite directions (iknown).

    Parameters
    ----------
    boundary_mask (array-like): The boundary mask for the lattice.

    Returns
    -------
    (tuple of array-like): The missing and known indices for the boundary condition.

    Notes
    -----
    This method calculates the missing and known indices based on the boundary mask. The missing indices are the
    non-zero entries of the boundary mask, and the known indices are their corresponding opposite directions.
    """

    # Find imissing, iknown 1-to-1 corresponding indices
    # Note: the "zero" index is used as default value here and won't affect BC computations
    nbd = len(self.indices[0])
    imissing = np.vstack([np.arange(self.lattice.q, dtype="uint8")] * nbd)
    iknown = np.vstack([self.lattice.opp_indices] * nbd)
    imissing[~boundary_mask] = 0
    iknown[~boundary_mask] = 0
    return imissing, iknown

get_missing_mask

get_missing_mask(boundary_mask)

Returns three boolean arrays the same shape as boundary_mask. Note: these boundary masks are useful for reduction (eg. summation) operators of selected q-directions.

Parameters

boundary_mask (array-like): The boundary mask for the lattice.

Returns

(tuple of array-like): The missing, known, and middle masks for the boundary condition.

Notes

This method calculates the missing, known, and middle masks based on the boundary mask. The missing mask is the boundary mask, the known mask is the opposite directions of the missing mask, and the middle mask is the directions that are neither missing nor known.

Source code in jax_lab/core/boundary_conditions.py
def get_missing_mask(self, boundary_mask):
    """
    Returns three boolean arrays the same shape as boundary_mask.
    Note: these boundary masks are useful for reduction (eg. summation) operators of selected q-directions.

    Parameters
    ----------
    boundary_mask (array-like): The boundary mask for the lattice.

    Returns
    -------
    (tuple of array-like): The missing, known, and middle masks for the boundary condition.

    Notes
    -----
    This method calculates the missing, known, and middle masks based on the boundary mask. The missing mask
    is the boundary mask, the known mask is the opposite directions of the missing mask, and the middle mask
    is the directions that are neither missing nor known.
    """
    # Find masks for imissing, iknown and imiddle
    imissingMask = boundary_mask
    iknownMask = imissingMask[:, self.lattice.opp_indices]
    imiddleMask = ~(imissingMask | iknownMask)
    return imissingMask, iknownMask, imiddleMask

get_normals

get_normals(boundary_mask)

Calculates the normal vectors at the boundary nodes.

Parameters

boundary_mask (array-like): The boundary mask for the lattice.

Returns

(array-like): The normal vectors at the boundary nodes.

Notes

This method calculates the normal vectors by dotting the boundary mask with the main lattice directions.

Source code in jax_lab/core/boundary_conditions.py
def get_normals(self, boundary_mask):
    """
    Calculates the normal vectors at the boundary nodes.

    Parameters
    ----------
    boundary_mask (array-like): The boundary mask for the lattice.

    Returns
    -------
    (array-like): The normal vectors at the boundary nodes.

    Notes
    -----
    This method calculates the normal vectors by dotting the boundary mask with the main lattice directions.
    """
    main_c = self.lattice.c.T[self.lattice.main_indices]
    m = boundary_mask[..., self.lattice.main_indices]
    normals = -np.dot(m, main_c)
    return normals

momentum_exchange_force

momentum_exchange_force(f_poststreaming, f_postcollision)

Using the momentum exchange method to compute the boundary force vector exerted on the solid geometry based on [1] as described in [3]. Ref [2] shows how [1] is applicable to curved geometries only by using a bounce-back method (e.g. Bouzidi) that accounts for curved boundaries. NOTE: this function should be called after BC’s are imposed. [1] A.J.C. Ladd, Numerical simulations of particular suspensions via a discretized Boltzmann equation. Part 2 (numerical results), J. Fluid Mech. 271 (1994) 311-339. [2] R. Mei, D. Yu, W. Shyy, L.-S. Luo, Force evaluation in the lattice Boltzmann method involving curved geometry, Phys. Rev. E 65 (2002) 041203. [3] Caiazzo, A., & Junk, M. (2008). Boundary forces in lattice Boltzmann: Analysis of momentum exchange algorithm. Computers & Mathematics with Applications, 55(7), 1415-1423.

Parameters

f_poststreaming (jax.numpy.ndarray): The post-streaming distribution function at each node in the lattice.

f_postcollision (jax.numpy.ndarray): The post-collision distribution function at each node in the lattice.

Returns

(jax.numpy.ndarray): The force exerted on the solid geometry at each boundary node.

Notes

This method computes the force exerted on the solid geometry at each boundary node using the momentum exchange method. The force is computed based on the post-streaming and post-collision distribution functions. This method should be called after the boundary conditions are imposed.

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0,))
def momentum_exchange_force(self, f_poststreaming, f_postcollision):
    """
    Using the momentum exchange method to compute the boundary force vector exerted on the solid geometry
    based on [1] as described in [3]. Ref [2] shows how [1] is applicable to curved geometries only by using a
    bounce-back method (e.g. Bouzidi) that accounts for curved boundaries.
    NOTE: this function should be called after BC's are imposed.
    [1] A.J.C. Ladd, Numerical simulations of particular suspensions via a discretized Boltzmann equation.
        Part 2 (numerical results), J. Fluid Mech. 271 (1994) 311-339.
    [2] R. Mei, D. Yu, W. Shyy, L.-S. Luo, Force evaluation in the lattice Boltzmann method involving
        curved geometry, Phys. Rev. E 65 (2002) 041203.
    [3] Caiazzo, A., & Junk, M. (2008). Boundary forces in lattice Boltzmann: Analysis of momentum exchange
        algorithm. Computers & Mathematics with Applications, 55(7), 1415-1423.

    Parameters
    ----------
    f_poststreaming (jax.numpy.ndarray): The post-streaming distribution function at each node in the lattice.

    f_postcollision (jax.numpy.ndarray): The post-collision distribution function at each node in the lattice.

    Returns
    -------
    (jax.numpy.ndarray): The force exerted on the solid geometry at each boundary node.

    Notes
    -----
    This method computes the force exerted on the solid geometry at each boundary node using the momentum exchange method.
    The force is computed based on the post-streaming and post-collision distribution functions. This method
    should be called after the boundary conditions are imposed.
    """
    c = jnp.array(self.lattice.c, dtype=self.precision_policy.compute_dtype)
    nbd = len(self.indices[0])
    bindex = np.arange(nbd)[:, None]
    phi = f_postcollision[self.indices][bindex, self.iknown] + f_poststreaming[self.indices][bindex, self.imissing]
    force = jnp.sum(c[:, self.iknown] * phi, axis=-1).T
    return force

momentum_flux

momentum_flux(fneq)

Compute the momentum flux.

Parameters

fneq (jax.numpy.ndarray): The non-equilibrium distribution function at each node in the lattice.

Returns

(jax.numpy.ndarray): The momentum flux at each node in the lattice.

Notes

This method computes the momentum flux by dotting the non-equilibrium distribution function with the lattice direction vectors.

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0,))
def momentum_flux(self, fneq):
    """
    Compute the momentum flux.

    Parameters
    ----------
    fneq (jax.numpy.ndarray): The non-equilibrium distribution function at each node in the lattice.

    Returns
    -------
    (jax.numpy.ndarray): The momentum flux at each node in the lattice.

    Notes
    -----
    This method computes the momentum flux by dotting the non-equilibrium distribution function with the lattice
    direction vectors.
    """
    return jnp.dot(fneq, self.lattice.cc)

prepare_populations

prepare_populations(fout, fin, implementation_step)

Prepares the distribution functions for the boundary condition.

Parameters

fout (jax.numpy.ndarray): The incoming distribution functions.

fin (jax.numpy.ndarray): The outgoing distribution functions.

implementation_step (str): The step in the lattice Boltzmann method algorithm at which the preparation is applied.

Returns

(jax.numpy.ndarray): The prepared distribution functions.

Notes

This method should be overridden in subclasses if the boundary condition requires preparation of the distribution functions during post-collision or post-streaming. See ExtrapolationBoundaryCondition for an example.

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0, 3), inline=True)
def prepare_populations(self, fout, fin, implementation_step):
    """
    Prepares the distribution functions for the boundary condition.

    Parameters
    ----------
    fout (jax.numpy.ndarray): The incoming distribution functions.

    fin (jax.numpy.ndarray): The outgoing distribution functions.

    implementation_step (str): The step in the lattice Boltzmann method algorithm at which the preparation is
        applied.

    Returns
    -------
    (jax.numpy.ndarray): The prepared distribution functions.

    Notes
    -----
    This method should be overridden in subclasses if the boundary condition requires preparation of the distribution functions during post-collision or post-streaming.
    See ExtrapolationBoundaryCondition for an example.
    """
    return fout

jax_lab.core.boundary_conditions.BounceBack

Bases: BoundaryCondition

Bounce-back boundary condition for a lattice Boltzmann method simulation.

This class implements a full-way bounce-back boundary condition, where particles hitting the boundary are reflected back in the direction they came from. The boundary condition is applied after the collision step.

Attributes

name (str): The name of the boundary condition. For this class, it is “BounceBackFullway”.

implementation_step (str): The step in the lattice Boltzmann method algorithm at which the boundary condition is applied. For this class, it is “PostCollision”.

theta (jax.numpy.ndarray; Default: None): Contact angle, applied for multiphase flows and only set for wall boundary conditions.

phi (jax.numpy.ndarray; Default: None): Contact angle parameter phi, applied for multiphase flows and only set for wall boundary conditions.

delta_rho (jax.numpy.ndarray; Default: None): Contact angle parameter delta_rho, applied for multiphase flows and only set for wall boundary conditions.

Source code in jax_lab/core/boundary_conditions.py
class BounceBack(BoundaryCondition):
    """
    Bounce-back boundary condition for a lattice Boltzmann method simulation.

    This class implements a full-way bounce-back boundary condition, where particles hitting the boundary are reflected
    back in the direction they came from. The boundary condition is applied after the collision step.

    Attributes
    ----------
    name (str): The name of the boundary condition. For this class, it is "BounceBackFullway".

    implementation_step (str): The step in the lattice Boltzmann method algorithm at which the boundary condition is applied. For this class, it is "PostCollision".

    theta (jax.numpy.ndarray; Default: None): Contact angle, applied for multiphase flows and only set for wall boundary conditions.

    phi (jax.numpy.ndarray; Default: None): Contact angle parameter phi, applied for multiphase flows and only set for wall boundary conditions.

    delta_rho (jax.numpy.ndarray; Default: None): Contact angle parameter delta_rho, applied for multiphase flows and only set for wall boundary conditions.
    """

    def __init__(self, indices, grid_info, precision_policy, theta=None, phi=None, delta_rho=None):
        super().__init__(indices, grid_info, precision_policy)
        self.name = "BounceBackFullway"
        self.implementation_step = "PostCollision"
        self.theta = theta
        self.phi = phi
        self.delta_rho = delta_rho

    @partial(jit, static_argnums=(0,))
    def apply(self, fout, fin):
        """
        Applies the bounce-back boundary condition.

        Parameters
        ----------
        fout (jax.numpy.ndarray): The output distribution functions.

        fin (jax.numpy.ndarray): The input distribution functions.

        Returns
        -------
        (jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.

        Notes
        -----
        This method applies the bounce-back boundary condition by reflecting the input distribution functions at the
        boundary nodes in the opposite direction.
        """

        return fin[self.indices][..., self.lattice.opp_indices]

apply

apply(fout, fin)

Applies the bounce-back boundary condition.

Parameters

fout (jax.numpy.ndarray): The output distribution functions.

fin (jax.numpy.ndarray): The input distribution functions.

Returns

(jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.

Notes

This method applies the bounce-back boundary condition by reflecting the input distribution functions at the boundary nodes in the opposite direction.

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0,))
def apply(self, fout, fin):
    """
    Applies the bounce-back boundary condition.

    Parameters
    ----------
    fout (jax.numpy.ndarray): The output distribution functions.

    fin (jax.numpy.ndarray): The input distribution functions.

    Returns
    -------
    (jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.

    Notes
    -----
    This method applies the bounce-back boundary condition by reflecting the input distribution functions at the
    boundary nodes in the opposite direction.
    """

    return fin[self.indices][..., self.lattice.opp_indices]

jax_lab.core.boundary_conditions.BounceBackMoving

Bases: BoundaryCondition

Moving bounce-back boundary condition for a lattice Boltzmann method simulation.

This class implements a moving bounce-back boundary condition, where particles hitting the boundary are reflected back in the direction they came from, with an additional velocity due to the movement of the boundary. The boundary condition is applied after the collision step.

Attributes

name (str): The name of the boundary condition. For this class, it is “BounceBackFullwayMoving”.

implementation_step (str): The step in the lattice Boltzmann method algorithm at which the boundary condition is applied. For this class, it is “PostCollision”.

is_dynamic (bool): Whether the boundary condition is dynamic (changes over time). For this class, it is True.

update_function (function): A function that updates the boundary condition. For this class, it is a function that updates the boundary condition based on the current time step. The signature of the function is update_function(time) -> (indices, vel),

theta (jax.numpy.ndarray; Default: None): Contact angle, applied for multiphase flows and only set for wall boundary conditions.

phi (jax.numpy.ndarray; Default: None): Contact angle parameter phi, applied for multiphase flows and only set for wall boundary conditions.

delta_rho (pytree of jax.numpy.ndarray; Default: None): Contact angle parameter delta_rho, applied for multiphase flows and only set for wall boundary conditions.

Source code in jax_lab/core/boundary_conditions.py
class BounceBackMoving(BoundaryCondition):
    """
    Moving bounce-back boundary condition for a lattice Boltzmann method simulation.

    This class implements a moving bounce-back boundary condition, where particles hitting the boundary are reflected
    back in the direction they came from, with an additional velocity due to the movement of the boundary. The boundary
    condition is applied after the collision step.

    Attributes
    ----------
    name (str): The name of the boundary condition. For this class, it is "BounceBackFullwayMoving".

    implementation_step (str): The step in the lattice Boltzmann method algorithm at which the boundary condition is applied. For this class, it is "PostCollision".

    is_dynamic (bool): Whether the boundary condition is dynamic (changes over time). For this class, it is True.

    update_function (function): A function that updates the boundary condition. For this class, it is a function that updates the boundary
    condition based on the current time step. The signature of the function is `update_function(time) -> (indices, vel)`,

    theta (jax.numpy.ndarray; Default: None): Contact angle, applied for multiphase flows and only set for wall boundary conditions.

    phi (jax.numpy.ndarray; Default: None): Contact angle parameter phi, applied for multiphase flows and only set for wall boundary conditions.

    delta_rho (pytree of jax.numpy.ndarray; Default: None): Contact angle parameter delta_rho, applied for multiphase flows and only set for wall boundary conditions.
    """

    def __init__(self, grid_info, precision_policy, update_function=None, theta=None, phi=None, delta_rho=None):
        # We get the indices at time zero to pass to the parent class for initialization
        indices, _ = update_function(0)
        super().__init__(indices, grid_info, precision_policy)
        self.name = "BounceBackFullwayMoving"
        self.implementation_step = "PostCollision"
        self.is_dynamic = True
        self.update_function = jit(update_function)
        self.theta = theta
        self.phi = phi
        self.delta_rho = delta_rho

    @partial(jit, static_argnums=(0,))
    def apply(self, fout, fin, time):
        """
        Applies the moving bounce-back boundary condition.

        Parameters
        ----------
        fout (jax.numpy.ndarray): The output distribution functions.

        fin (jax.numpy.ndarray): The input distribution functions.

        time (int): The current time step.

        Returns
        -------
        (jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.
        """
        indices, vel = self.update_function(time)
        c = jnp.array(self.lattice.c, dtype=self.precision_policy.compute_dtype)
        cu = 6.0 * self.lattice.w * jnp.dot(vel, c)
        return fout.at[indices].set(fin[indices][..., self.lattice.opp_indices] - cu)

apply

apply(fout, fin, time)

Applies the moving bounce-back boundary condition.

Parameters

fout (jax.numpy.ndarray): The output distribution functions.

fin (jax.numpy.ndarray): The input distribution functions.

time (int): The current time step.

Returns

(jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0,))
def apply(self, fout, fin, time):
    """
    Applies the moving bounce-back boundary condition.

    Parameters
    ----------
    fout (jax.numpy.ndarray): The output distribution functions.

    fin (jax.numpy.ndarray): The input distribution functions.

    time (int): The current time step.

    Returns
    -------
    (jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.
    """
    indices, vel = self.update_function(time)
    c = jnp.array(self.lattice.c, dtype=self.precision_policy.compute_dtype)
    cu = 6.0 * self.lattice.w * jnp.dot(vel, c)
    return fout.at[indices].set(fin[indices][..., self.lattice.opp_indices] - cu)

jax_lab.core.boundary_conditions.BounceBackHalfway

Bases: BoundaryCondition

Halfway bounce-back boundary condition for a lattice Boltzmann method simulation.

This class implements a halfway bounce-back boundary condition. The boundary condition is applied after the streaming step.

Attributes

name (str): The name of the boundary condition. For this class, it is “BounceBackHalfway”.

implementation_step (str): The step in the lattice Boltzmann method algorithm at which the boundary condition is applied. For this class, it is “PostStreaming”.

needs_extra_configuration (bool): Whether the boundary condition needs extra configuration before it can be applied. For this class, it is True.

is_solid (bool): Whether the boundary condition represents a solid boundary. For this class, it is True.

solid_indices (tuple): Original solid-node indices, stored by configure before self.indices is shifted to the adjacent fluid nodes. Used by multiphase wetting schemes.

vel (array-like): The prescribed value of velocity vector for the boundary condition. No-slip BC is assumed if vel=None (default).

theta (jax.numpy.ndarray; Default: None): Contact angle, applied for multiphase flows and only set for wall boundary conditions.

phi (jax.numpy.ndarray; Default: None): Contact angle parameter phi, applied for multiphase flows and only set for wall boundary conditions.

delta_rho (pytree of jax.numpy.ndarray; Default: None): Contact angle parameter delta_rho, applied for multiphase flows and only set for wall boundary conditions.

Source code in jax_lab/core/boundary_conditions.py
class BounceBackHalfway(BoundaryCondition):
    """
    Halfway bounce-back boundary condition for a lattice Boltzmann method simulation.

    This class implements a halfway bounce-back boundary condition. The boundary condition is applied after
    the streaming step.

    Attributes
    ----------
    name (str): The name of the boundary condition. For this class, it is "BounceBackHalfway".

    implementation_step (str): The step in the lattice Boltzmann method algorithm at which the boundary condition is applied. For this class, it is "PostStreaming".

    needs_extra_configuration (bool): Whether the boundary condition needs extra configuration before it can be applied. For this class, it is True.

    is_solid (bool): Whether the boundary condition represents a solid boundary. For this class, it is True.

    solid_indices (tuple): Original solid-node indices, stored by configure before self.indices is shifted to the
    adjacent fluid nodes. Used by multiphase wetting schemes.

    vel (array-like): The prescribed value of velocity vector for the boundary condition. No-slip BC is assumed if vel=None (default).

    theta (jax.numpy.ndarray; Default: None): Contact angle, applied for multiphase flows and only set for wall boundary conditions.

    phi (jax.numpy.ndarray; Default: None): Contact angle parameter phi, applied for multiphase flows and only set for wall boundary conditions.

    delta_rho (pytree of jax.numpy.ndarray; Default: None): Contact angle parameter delta_rho, applied for multiphase flows and only set for wall boundary conditions.
    """

    def __init__(self, indices, grid_info, precision_policy, vel=None, theta=None, phi=None, delta_rho=None):
        super().__init__(indices, grid_info, precision_policy)
        self.name = "BounceBackHalfway"
        self.implementation_step = "PostStreaming"
        self.needs_extra_configuration = True
        self.is_solid = True
        self.vel = vel
        self.theta = theta
        self.phi = phi
        self.delta_rho = delta_rho

    def configure(self, boundary_mask):
        """
        Configures the boundary condition.

        Parameters
        ----------
        boundary_mask (array-like): The grid mask for the boundary voxels.

        Returns
        -------
        None

        Notes
        -----
        This method performs an index shift for the halfway bounce-back boundary condition. It updates the indices of
        the boundary nodes to be the indices of fluid nodes adjacent of the solid nodes.
        """
        # Keep the original solid-node indices; multiphase wetting schemes need them
        # after self.indices is shifted to the adjacent fluid nodes below.
        self.solid_indices = self.indices
        # Perform index shift for halfway BB.
        hasFluidNeighbour = ~boundary_mask[:, self.lattice.opp_indices]
        nbd_orig = len(self.indices[0])
        idx = np.array(self.indices).T
        idx_trg = []
        for i in range(self.lattice.q):
            idx_trg.append(idx[hasFluidNeighbour[:, i], :] + self.lattice.c[:, i])
        indices_new = np.unique(np.vstack(idx_trg), axis=0)
        self.indices = tuple(indices_new.T)
        nbd_modified = len(self.indices[0])
        if (nbd_orig != nbd_modified) and self.vel is not None:
            vel_avg = np.mean(self.vel, axis=0)
            self.vel = jnp.zeros(indices_new.shape, dtype=self.precision_policy.compute_dtype) + vel_avg
            warnings.warn(
                "A constant averaged velocity vector is imposed at all boundary cells.",
                UserWarning,
                stacklevel=2,
            )

        return

    @partial(jit, static_argnums=(0, 3), inline=True)
    def prepare_populations(self, fout, fin, implementation_step):
        """
        Pin solid-node populations to the rest equilibrium (rho = 1, u = 0).

        Halfway bounce-back never constrains the populations at the solid nodes, so they evolve freely and can diverge.
        This is harmless in single phase simulations but multiphase wetting schemes read wall densities, so the solid
        nodes are reset after every streaming step. The wetted wall density used in the force computation is overwritten
        by apply_contact_angle, so the pinned value never enters the wetting force directly.

        Parameters
        ----------
        fout (jax.numpy.ndarray): The post-streaming or post-collision distribution functions.

        fin (jax.numpy.ndarray): The pre-collision or post-collision distribution functions.

        implementation_step (str): The step at which the preparation is applied.

        Returns
        -------
        (jax.numpy.ndarray): The distribution functions with solid nodes reset after streaming.
        """
        if implementation_step == "PostStreaming":
            return fout.at[self.solid_indices].set(self.precision_policy.cast_to_output(self.lattice.w))
        return fout

    @staticmethod
    def reflect_missing(fbd, fin_bd, imissing, iknown):
        """
        Core halfway bounce-back math: set each missing direction to the post-streaming value coming from its
        known (opposite) direction. Pure function of its arguments (no BC-instance state), so it is reusable
        both by apply() (global indices) and by a local (per-shard) bounce-back kernel operating on a gathered
        block with the same (n, q) shape.

        Parameters
        ----------
        fbd (jax.numpy.ndarray): Post-streaming distribution functions at the boundary nodes, shape (n, q).

        fin_bd (jax.numpy.ndarray): Post-streaming distribution functions at the same boundary nodes, shape (n, q).

        imissing (array-like): Missing-direction indices, shape (n, q).

        iknown (array-like): Known (opposite) direction indices, shape (n, q).

        Returns
        -------
        (jax.numpy.ndarray): fbd with each missing direction set from its known direction.
        """
        bindex = jnp.arange(fbd.shape[0])[:, None]
        return fbd.at[bindex, imissing].set(fin_bd[bindex, iknown])

    @staticmethod
    def velocity_correction(fbd, imissing, iknown, vel, w, c):
        """
        Core halfway bounce-back velocity-forcing math, factored out of impose_boundary_vel so it is reusable by
        a local (per-shard) bounce-back kernel. Pure function of its arguments; vel=0 is a safe no-op, so a
        local kernel merging several boundary conditions can always call this instead of conditionally skipping
        it per boundary condition.

        Parameters
        ----------
        fbd (jax.numpy.ndarray): Distribution functions at the boundary nodes, shape (n, q).

        imissing (array-like): Missing-direction indices, shape (n, q).

        iknown (array-like): Known (opposite) direction indices, shape (n, q).

        vel (jax.numpy.ndarray): Prescribed velocity vector, shape (n, dim).

        w (jax.numpy.ndarray): Lattice weights, shape (q,).

        c (jax.numpy.ndarray): Lattice velocity vectors, shape (dim, q).

        Returns
        -------
        (jax.numpy.ndarray): fbd with the velocity correction applied.
        """
        bindex = jnp.arange(fbd.shape[0])[:, None]
        cu = 6.0 * w * jnp.dot(vel, c)
        return fbd.at[bindex, imissing].add(-cu[bindex, iknown])

    @partial(jit, static_argnums=(0,))
    def impose_boundary_vel(self, fbd, bindex):
        c = jnp.array(self.lattice.c, dtype=self.precision_policy.compute_dtype)
        return self.velocity_correction(fbd, self.imissing, self.iknown, self.vel, self.lattice.w, c)

    @partial(jit, static_argnums=(0,))
    def apply(self, fout, fin):
        """
        Applies the halfway bounce-back boundary condition.

        Parameters
        ----------
        fout (jax.numpy.ndarray): The output distribution functions.

        fin (jax.numpy.ndarray): The input distribution functions.

        Returns
        -------
        (jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.
        """
        fbd = self.reflect_missing(fout[self.indices], fin[self.indices], self.imissing, self.iknown)
        if self.vel is not None:
            bindex = np.arange(len(self.indices[0]))[:, None]
            fbd = self.impose_boundary_vel(fbd, bindex)
        return fbd

apply

apply(fout, fin)

Applies the halfway bounce-back boundary condition.

Parameters

fout (jax.numpy.ndarray): The output distribution functions.

fin (jax.numpy.ndarray): The input distribution functions.

Returns

(jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0,))
def apply(self, fout, fin):
    """
    Applies the halfway bounce-back boundary condition.

    Parameters
    ----------
    fout (jax.numpy.ndarray): The output distribution functions.

    fin (jax.numpy.ndarray): The input distribution functions.

    Returns
    -------
    (jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.
    """
    fbd = self.reflect_missing(fout[self.indices], fin[self.indices], self.imissing, self.iknown)
    if self.vel is not None:
        bindex = np.arange(len(self.indices[0]))[:, None]
        fbd = self.impose_boundary_vel(fbd, bindex)
    return fbd

configure

configure(boundary_mask)

Configures the boundary condition.

Parameters

boundary_mask (array-like): The grid mask for the boundary voxels.

Returns

None

Notes

This method performs an index shift for the halfway bounce-back boundary condition. It updates the indices of the boundary nodes to be the indices of fluid nodes adjacent of the solid nodes.

Source code in jax_lab/core/boundary_conditions.py
def configure(self, boundary_mask):
    """
    Configures the boundary condition.

    Parameters
    ----------
    boundary_mask (array-like): The grid mask for the boundary voxels.

    Returns
    -------
    None

    Notes
    -----
    This method performs an index shift for the halfway bounce-back boundary condition. It updates the indices of
    the boundary nodes to be the indices of fluid nodes adjacent of the solid nodes.
    """
    # Keep the original solid-node indices; multiphase wetting schemes need them
    # after self.indices is shifted to the adjacent fluid nodes below.
    self.solid_indices = self.indices
    # Perform index shift for halfway BB.
    hasFluidNeighbour = ~boundary_mask[:, self.lattice.opp_indices]
    nbd_orig = len(self.indices[0])
    idx = np.array(self.indices).T
    idx_trg = []
    for i in range(self.lattice.q):
        idx_trg.append(idx[hasFluidNeighbour[:, i], :] + self.lattice.c[:, i])
    indices_new = np.unique(np.vstack(idx_trg), axis=0)
    self.indices = tuple(indices_new.T)
    nbd_modified = len(self.indices[0])
    if (nbd_orig != nbd_modified) and self.vel is not None:
        vel_avg = np.mean(self.vel, axis=0)
        self.vel = jnp.zeros(indices_new.shape, dtype=self.precision_policy.compute_dtype) + vel_avg
        warnings.warn(
            "A constant averaged velocity vector is imposed at all boundary cells.",
            UserWarning,
            stacklevel=2,
        )

    return

prepare_populations

prepare_populations(fout, fin, implementation_step)

Pin solid-node populations to the rest equilibrium (rho = 1, u = 0).

Halfway bounce-back never constrains the populations at the solid nodes, so they evolve freely and can diverge. This is harmless in single phase simulations but multiphase wetting schemes read wall densities, so the solid nodes are reset after every streaming step. The wetted wall density used in the force computation is overwritten by apply_contact_angle, so the pinned value never enters the wetting force directly.

Parameters

fout (jax.numpy.ndarray): The post-streaming or post-collision distribution functions.

fin (jax.numpy.ndarray): The pre-collision or post-collision distribution functions.

implementation_step (str): The step at which the preparation is applied.

Returns

(jax.numpy.ndarray): The distribution functions with solid nodes reset after streaming.

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0, 3), inline=True)
def prepare_populations(self, fout, fin, implementation_step):
    """
    Pin solid-node populations to the rest equilibrium (rho = 1, u = 0).

    Halfway bounce-back never constrains the populations at the solid nodes, so they evolve freely and can diverge.
    This is harmless in single phase simulations but multiphase wetting schemes read wall densities, so the solid
    nodes are reset after every streaming step. The wetted wall density used in the force computation is overwritten
    by apply_contact_angle, so the pinned value never enters the wetting force directly.

    Parameters
    ----------
    fout (jax.numpy.ndarray): The post-streaming or post-collision distribution functions.

    fin (jax.numpy.ndarray): The pre-collision or post-collision distribution functions.

    implementation_step (str): The step at which the preparation is applied.

    Returns
    -------
    (jax.numpy.ndarray): The distribution functions with solid nodes reset after streaming.
    """
    if implementation_step == "PostStreaming":
        return fout.at[self.solid_indices].set(self.precision_policy.cast_to_output(self.lattice.w))
    return fout

reflect_missing staticmethod

reflect_missing(fbd, fin_bd, imissing, iknown)

Core halfway bounce-back math: set each missing direction to the post-streaming value coming from its known (opposite) direction. Pure function of its arguments (no BC-instance state), so it is reusable both by apply() (global indices) and by a local (per-shard) bounce-back kernel operating on a gathered block with the same (n, q) shape.

Parameters

fbd (jax.numpy.ndarray): Post-streaming distribution functions at the boundary nodes, shape (n, q).

fin_bd (jax.numpy.ndarray): Post-streaming distribution functions at the same boundary nodes, shape (n, q).

imissing (array-like): Missing-direction indices, shape (n, q).

iknown (array-like): Known (opposite) direction indices, shape (n, q).

Returns

(jax.numpy.ndarray): fbd with each missing direction set from its known direction.

Source code in jax_lab/core/boundary_conditions.py
@staticmethod
def reflect_missing(fbd, fin_bd, imissing, iknown):
    """
    Core halfway bounce-back math: set each missing direction to the post-streaming value coming from its
    known (opposite) direction. Pure function of its arguments (no BC-instance state), so it is reusable
    both by apply() (global indices) and by a local (per-shard) bounce-back kernel operating on a gathered
    block with the same (n, q) shape.

    Parameters
    ----------
    fbd (jax.numpy.ndarray): Post-streaming distribution functions at the boundary nodes, shape (n, q).

    fin_bd (jax.numpy.ndarray): Post-streaming distribution functions at the same boundary nodes, shape (n, q).

    imissing (array-like): Missing-direction indices, shape (n, q).

    iknown (array-like): Known (opposite) direction indices, shape (n, q).

    Returns
    -------
    (jax.numpy.ndarray): fbd with each missing direction set from its known direction.
    """
    bindex = jnp.arange(fbd.shape[0])[:, None]
    return fbd.at[bindex, imissing].set(fin_bd[bindex, iknown])

velocity_correction staticmethod

velocity_correction(fbd, imissing, iknown, vel, w, c)

Core halfway bounce-back velocity-forcing math, factored out of impose_boundary_vel so it is reusable by a local (per-shard) bounce-back kernel. Pure function of its arguments; vel=0 is a safe no-op, so a local kernel merging several boundary conditions can always call this instead of conditionally skipping it per boundary condition.

Parameters

fbd (jax.numpy.ndarray): Distribution functions at the boundary nodes, shape (n, q).

imissing (array-like): Missing-direction indices, shape (n, q).

iknown (array-like): Known (opposite) direction indices, shape (n, q).

vel (jax.numpy.ndarray): Prescribed velocity vector, shape (n, dim).

w (jax.numpy.ndarray): Lattice weights, shape (q,).

c (jax.numpy.ndarray): Lattice velocity vectors, shape (dim, q).

Returns

(jax.numpy.ndarray): fbd with the velocity correction applied.

Source code in jax_lab/core/boundary_conditions.py
@staticmethod
def velocity_correction(fbd, imissing, iknown, vel, w, c):
    """
    Core halfway bounce-back velocity-forcing math, factored out of impose_boundary_vel so it is reusable by
    a local (per-shard) bounce-back kernel. Pure function of its arguments; vel=0 is a safe no-op, so a
    local kernel merging several boundary conditions can always call this instead of conditionally skipping
    it per boundary condition.

    Parameters
    ----------
    fbd (jax.numpy.ndarray): Distribution functions at the boundary nodes, shape (n, q).

    imissing (array-like): Missing-direction indices, shape (n, q).

    iknown (array-like): Known (opposite) direction indices, shape (n, q).

    vel (jax.numpy.ndarray): Prescribed velocity vector, shape (n, dim).

    w (jax.numpy.ndarray): Lattice weights, shape (q,).

    c (jax.numpy.ndarray): Lattice velocity vectors, shape (dim, q).

    Returns
    -------
    (jax.numpy.ndarray): fbd with the velocity correction applied.
    """
    bindex = jnp.arange(fbd.shape[0])[:, None]
    cu = 6.0 * w * jnp.dot(vel, c)
    return fbd.at[bindex, imissing].add(-cu[bindex, iknown])

jax_lab.core.boundary_conditions.EquilibriumBC

Bases: BoundaryCondition

Equilibrium boundary condition for a lattice Boltzmann method simulation.

This class implements an equilibrium boundary condition, where the distribution function at the boundary nodes is set to the equilibrium distribution function. The boundary condition is applied after the streaming step.

Attributes

name (str): The name of the boundary condition. For this class, it is “EquilibriumBC”.

implementation_step (str): The step in the lattice Boltzmann method algorithm at which the boundary condition is applied. For this class, it is “PostStreaming”.

out (jax.numpy.ndarray): The equilibrium distribution function at the boundary nodes.

Source code in jax_lab/core/boundary_conditions.py
class EquilibriumBC(BoundaryCondition):
    """
    Equilibrium boundary condition for a lattice Boltzmann method simulation.

    This class implements an equilibrium boundary condition, where the distribution function at the boundary nodes is
    set to the equilibrium distribution function. The boundary condition is applied after the streaming step.

    Attributes
    ----------
    name (str): The name of the boundary condition. For this class, it is "EquilibriumBC".

    implementation_step (str): The step in the lattice Boltzmann method algorithm at which the boundary condition is applied. For this class, it is "PostStreaming".

    out (jax.numpy.ndarray): The equilibrium distribution function at the boundary nodes.
    """

    def __init__(self, indices, grid_info, precision_policy, rho, u):
        super().__init__(indices, grid_info, precision_policy)
        self.out = self.precision_policy.cast_to_output(self.equilibrium(rho, u))
        self.name = "EquilibriumBC"
        self.implementation_step = "PostStreaming"

    @partial(jit, static_argnums=(0,))
    def apply(self, fout, fin):
        """
        Applies the equilibrium boundary condition.

        Parameters
        ----------
        fout (jax.numpy.ndarray): The output distribution functions.

        fin (jax.numpy.ndarray): The input distribution functions.

        Returns
        -------
        (jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.

        Notes
        -----
        This method applies the equilibrium boundary condition by setting the output distribution functions at the
        boundary nodes to the equilibrium distribution function.
        """
        return self.out

apply

apply(fout, fin)

Applies the equilibrium boundary condition.

Parameters

fout (jax.numpy.ndarray): The output distribution functions.

fin (jax.numpy.ndarray): The input distribution functions.

Returns

(jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.

Notes

This method applies the equilibrium boundary condition by setting the output distribution functions at the boundary nodes to the equilibrium distribution function.

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0,))
def apply(self, fout, fin):
    """
    Applies the equilibrium boundary condition.

    Parameters
    ----------
    fout (jax.numpy.ndarray): The output distribution functions.

    fin (jax.numpy.ndarray): The input distribution functions.

    Returns
    -------
    (jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.

    Notes
    -----
    This method applies the equilibrium boundary condition by setting the output distribution functions at the
    boundary nodes to the equilibrium distribution function.
    """
    return self.out

jax_lab.core.boundary_conditions.DoNothing

Bases: BoundaryCondition

Preserve post-collision populations at selected boundary nodes.

Parameters

indices (tuple of numpy.ndarray): Boundary-node indices.

grid_info (dict): Grid and lattice metadata.

precision_policy (PrecisionPolicy): Compute and output precision policy.

Notes

Streaming is skipped at these nodes by returning the post-collision populations after streaming. This avoids values wrapping into the domain from the opposite side of the rolled population array.

Source code in jax_lab/core/boundary_conditions.py
class DoNothing(BoundaryCondition):
    """
    Preserve post-collision populations at selected boundary nodes.

    Parameters
    ----------
    indices (tuple of numpy.ndarray): Boundary-node indices.

    grid_info (dict): Grid and lattice metadata.

    precision_policy (PrecisionPolicy): Compute and output precision policy.

    Notes
    -----
    Streaming is skipped at these nodes by returning the post-collision
    populations after streaming. This avoids values wrapping into the domain
    from the opposite side of the rolled population array.
    """

    def __init__(self, indices, grid_info, precision_policy):
        super().__init__(indices, grid_info, precision_policy)
        self.name = "DoNothing"
        self.implementation_step = "PostStreaming"

    @partial(jit, static_argnums=(0,))
    def apply(self, fout, fin):
        """
        Applies the do-nothing boundary condition.

        Parameters
        ----------
        fout (jax.numpy.ndarray): The output distribution functions.

        fin (jax.numpy.ndarray): The input distribution functions.

        Returns
        -------
        jax.numpy.ndarray
            The modified output distribution functions after applying the boundary condition.

        Notes
        -----
        This method applies the do-nothing boundary condition by simply returning the input distribution functions at the
        boundary nodes.
        """
        return fin[self.indices]

apply

apply(fout, fin)

Applies the do-nothing boundary condition.

Parameters

fout (jax.numpy.ndarray): The output distribution functions.

fin (jax.numpy.ndarray): The input distribution functions.

Returns

jax.numpy.ndarray The modified output distribution functions after applying the boundary condition.

Notes

This method applies the do-nothing boundary condition by simply returning the input distribution functions at the boundary nodes.

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0,))
def apply(self, fout, fin):
    """
    Applies the do-nothing boundary condition.

    Parameters
    ----------
    fout (jax.numpy.ndarray): The output distribution functions.

    fin (jax.numpy.ndarray): The input distribution functions.

    Returns
    -------
    jax.numpy.ndarray
        The modified output distribution functions after applying the boundary condition.

    Notes
    -----
    This method applies the do-nothing boundary condition by simply returning the input distribution functions at the
    boundary nodes.
    """
    return fin[self.indices]

jax_lab.core.boundary_conditions.ZouHe

Bases: BoundaryCondition

Zou-He boundary condition for a lattice Boltzmann method simulation.

This class implements the Zou-He boundary condition, which is a non-equilibrium bounce-back boundary condition. It can be used to set inflow and outflow boundary conditions with prescribed pressure or velocity.

Attributes

name (str): The name of the boundary condition. For this class, it is “ZouHe”.

implementation_step (str): The step in the lattice Boltzmann method algorithm at which the boundary condition is applied. For this class,it is “PostStreaming”.

type (str): The type of the boundary condition. It can be either ‘velocity’ for a prescribed velocity boundary condition, or ‘pressure’ for a prescribed pressure boundary condition.

prescribed (float or array-like): The prescribed values for the boundary condition. It can be either the prescribed velocities for a ‘velocity’

boundary condition, or the prescribed pressures for a ‘pressure’ boundary condition.

References

Zou, Q., & He, X. (1997). On pressure and velocity boundary conditions for the lattice Boltzmann BGK model. Physics of Fluids, 9(6), 1591-1598. doi:10.1063/1.869307

Source code in jax_lab/core/boundary_conditions.py
class ZouHe(BoundaryCondition):
    """
    Zou-He boundary condition for a lattice Boltzmann method simulation.

    This class implements the Zou-He boundary condition, which is a non-equilibrium bounce-back boundary condition.
    It can be used to set inflow and outflow boundary conditions with prescribed pressure or velocity.

    Attributes
    ----------
    name (str): The name of the boundary condition. For this class, it is "ZouHe".

    implementation_step (str): The step in the lattice Boltzmann method algorithm at which the boundary condition is applied. For this class,it is "PostStreaming".

    type (str): The type of the boundary condition. It can be either 'velocity' for a prescribed velocity boundary condition,
    or 'pressure' for a prescribed pressure boundary condition.

    prescribed (float or array-like): The prescribed values for the boundary condition. It can be either the prescribed velocities for a 'velocity'

    boundary condition, or the prescribed pressures for a 'pressure' boundary condition.

    References
    ----------
    Zou, Q., & He, X. (1997). On pressure and velocity boundary conditions for the lattice Boltzmann BGK model.
    Physics of Fluids, 9(6), 1591-1598. doi:10.1063/1.869307
    """

    def __init__(self, indices, grid_info, precision_policy, ttype, prescribed):
        super().__init__(indices, grid_info, precision_policy)
        self.name = "ZouHe"
        self.implementation_step = "PostStreaming"
        self.type = ttype
        self.prescribed = prescribed
        self.needs_extra_configuration = True

    def configure(self, boundary_mask):
        """
        Correct boundary indices to ensure that only voxelized surfaces with normal vectors along main cartesian axes
        are assigned this type of BC.
        """
        nv = np.dot(self.lattice.c, ~boundary_mask.T)
        corner_voxels = np.count_nonzero(nv, axis=0) > 1
        # removed_voxels = np.array(self.indices)[:, corner_voxels]
        self.indices = tuple(np.array(self.indices)[:, ~corner_voxels])
        self.prescribed = self.prescribed[~corner_voxels]
        return

    @partial(jit, static_argnums=(0,), inline=True)
    def calculate_vel(self, fpop, rho):
        """
        Calculate velocity based on the prescribed pressure/density (Zou/He BC)
        """
        unormal = -1.0 + 1.0 / rho * (
            jnp.sum(fpop[self.indices] * self.imiddle_mask, axis=1, keepdims=True)
            + 2.0 * jnp.sum(fpop[self.indices] * self.iknown_mask, axis=1, keepdims=True)
        )

        # Return the above unormal as a normal vector which sets the tangential velocities to zero
        vel = unormal * self.normals
        return vel

    @partial(jit, static_argnums=(0,), inline=True)
    def calculate_rho(self, fpop, vel):
        """
        Calculate density based on the prescribed velocity (Zou/He BC)
        """
        unormal = np.sum(self.normals * vel, axis=1)

        rho = (1.0 / (1.0 + unormal))[..., None] * (
            jnp.sum(fpop[self.indices] * self.imiddle_mask, axis=1, keepdims=True)
            + 2.0 * jnp.sum(fpop[self.indices] * self.iknown_mask, axis=1, keepdims=True)
        )
        return rho

    @partial(jit, static_argnums=(0,), inline=True)
    def calculate_equilibrium(self, fpop):
        """
        This is the ZouHe method of calculating the missing macroscopic variables at the boundary.
        """
        if self.type == "velocity":
            vel = self.prescribed
            rho = self.calculate_rho(fpop, vel)
        elif self.type == "pressure":
            rho = self.prescribed
            vel = self.calculate_vel(fpop, rho)
        else:
            raise ValueError(f"type = {self.type} not supported! Use 'pressure' or 'velocity'.")

        # compute feq at the boundary
        feq = self.equilibrium(rho, vel)
        return feq

    @partial(jit, static_argnums=(0,), inline=True)
    def bounceback_nonequilibrium(self, fpop, feq):
        """
        Calculate unknown populations using bounce-back of non-equilibrium populations
        a la original Zou & He formulation
        """
        nbd = len(self.indices[0])
        bindex = np.arange(nbd)[:, None]
        fbd = fpop[self.indices]
        fknown = fpop[self.indices][bindex, self.iknown] + feq[bindex, self.imissing] - feq[bindex, self.iknown]
        # feq may be at compute precision while fbd (from fpop, i.e. fout) is at output precision (e.g. "f32/f16"):
        # cast explicitly instead of relying on an implicit narrowing cast in .set(), which JAX warns will become
        # an error in a future release.
        fbd = fbd.at[bindex, self.imissing].set(fknown.astype(fbd.dtype))
        return fbd

    @partial(jit, static_argnums=(0,))
    def apply(self, fout, _):
        """
        Applies the Zou-He boundary condition.

        Parameters
        ----------
        fout (jax.numpy.ndarray): The output distribution functions.

        _ (jax.numpy.ndarray): The input distribution functions. This is not used in this method.

        Returns
        -------
        (jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.

        Notes
        -----
        This method applies the Zou-He boundary condition by first computing the equilibrium distribution functions based
        on the prescribed values and the type of boundary condition, and then setting the unknown distribution functions
        based on the non-equilibrium bounce-back method.
        Tangential velocity is not ensured to be zero by adding transverse contributions based on
        Hecth & Harting (2010) (doi:10.1088/1742-5468/2010/01/P01018) as it caused numerical instabilities at higher
        Reynolds numbers. One needs to use "Regularized" BC at higher Reynolds.
        """
        # compute the equilibrium based on prescribed values and the type of BC
        feq = self.calculate_equilibrium(fout)

        # set the unknown f populations based on the non-equilibrium bounce-back method
        fbd = self.bounceback_nonequilibrium(fout, feq)

        return fbd

apply

apply(fout, _)

Applies the Zou-He boundary condition.

Parameters

fout (jax.numpy.ndarray): The output distribution functions.

_ (jax.numpy.ndarray): The input distribution functions. This is not used in this method.

Returns

(jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.

Notes

This method applies the Zou-He boundary condition by first computing the equilibrium distribution functions based on the prescribed values and the type of boundary condition, and then setting the unknown distribution functions based on the non-equilibrium bounce-back method. Tangential velocity is not ensured to be zero by adding transverse contributions based on Hecth & Harting (2010) (doi:10.1088/1742-5468/2010/01/P01018) as it caused numerical instabilities at higher Reynolds numbers. One needs to use “Regularized” BC at higher Reynolds.

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0,))
def apply(self, fout, _):
    """
    Applies the Zou-He boundary condition.

    Parameters
    ----------
    fout (jax.numpy.ndarray): The output distribution functions.

    _ (jax.numpy.ndarray): The input distribution functions. This is not used in this method.

    Returns
    -------
    (jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.

    Notes
    -----
    This method applies the Zou-He boundary condition by first computing the equilibrium distribution functions based
    on the prescribed values and the type of boundary condition, and then setting the unknown distribution functions
    based on the non-equilibrium bounce-back method.
    Tangential velocity is not ensured to be zero by adding transverse contributions based on
    Hecth & Harting (2010) (doi:10.1088/1742-5468/2010/01/P01018) as it caused numerical instabilities at higher
    Reynolds numbers. One needs to use "Regularized" BC at higher Reynolds.
    """
    # compute the equilibrium based on prescribed values and the type of BC
    feq = self.calculate_equilibrium(fout)

    # set the unknown f populations based on the non-equilibrium bounce-back method
    fbd = self.bounceback_nonequilibrium(fout, feq)

    return fbd

bounceback_nonequilibrium

bounceback_nonequilibrium(fpop, feq)

Calculate unknown populations using bounce-back of non-equilibrium populations a la original Zou & He formulation

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0,), inline=True)
def bounceback_nonequilibrium(self, fpop, feq):
    """
    Calculate unknown populations using bounce-back of non-equilibrium populations
    a la original Zou & He formulation
    """
    nbd = len(self.indices[0])
    bindex = np.arange(nbd)[:, None]
    fbd = fpop[self.indices]
    fknown = fpop[self.indices][bindex, self.iknown] + feq[bindex, self.imissing] - feq[bindex, self.iknown]
    # feq may be at compute precision while fbd (from fpop, i.e. fout) is at output precision (e.g. "f32/f16"):
    # cast explicitly instead of relying on an implicit narrowing cast in .set(), which JAX warns will become
    # an error in a future release.
    fbd = fbd.at[bindex, self.imissing].set(fknown.astype(fbd.dtype))
    return fbd

calculate_equilibrium

calculate_equilibrium(fpop)

This is the ZouHe method of calculating the missing macroscopic variables at the boundary.

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0,), inline=True)
def calculate_equilibrium(self, fpop):
    """
    This is the ZouHe method of calculating the missing macroscopic variables at the boundary.
    """
    if self.type == "velocity":
        vel = self.prescribed
        rho = self.calculate_rho(fpop, vel)
    elif self.type == "pressure":
        rho = self.prescribed
        vel = self.calculate_vel(fpop, rho)
    else:
        raise ValueError(f"type = {self.type} not supported! Use 'pressure' or 'velocity'.")

    # compute feq at the boundary
    feq = self.equilibrium(rho, vel)
    return feq

calculate_rho

calculate_rho(fpop, vel)

Calculate density based on the prescribed velocity (Zou/He BC)

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0,), inline=True)
def calculate_rho(self, fpop, vel):
    """
    Calculate density based on the prescribed velocity (Zou/He BC)
    """
    unormal = np.sum(self.normals * vel, axis=1)

    rho = (1.0 / (1.0 + unormal))[..., None] * (
        jnp.sum(fpop[self.indices] * self.imiddle_mask, axis=1, keepdims=True)
        + 2.0 * jnp.sum(fpop[self.indices] * self.iknown_mask, axis=1, keepdims=True)
    )
    return rho

calculate_vel

calculate_vel(fpop, rho)

Calculate velocity based on the prescribed pressure/density (Zou/He BC)

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0,), inline=True)
def calculate_vel(self, fpop, rho):
    """
    Calculate velocity based on the prescribed pressure/density (Zou/He BC)
    """
    unormal = -1.0 + 1.0 / rho * (
        jnp.sum(fpop[self.indices] * self.imiddle_mask, axis=1, keepdims=True)
        + 2.0 * jnp.sum(fpop[self.indices] * self.iknown_mask, axis=1, keepdims=True)
    )

    # Return the above unormal as a normal vector which sets the tangential velocities to zero
    vel = unormal * self.normals
    return vel

configure

configure(boundary_mask)

Correct boundary indices to ensure that only voxelized surfaces with normal vectors along main cartesian axes are assigned this type of BC.

Source code in jax_lab/core/boundary_conditions.py
def configure(self, boundary_mask):
    """
    Correct boundary indices to ensure that only voxelized surfaces with normal vectors along main cartesian axes
    are assigned this type of BC.
    """
    nv = np.dot(self.lattice.c, ~boundary_mask.T)
    corner_voxels = np.count_nonzero(nv, axis=0) > 1
    # removed_voxels = np.array(self.indices)[:, corner_voxels]
    self.indices = tuple(np.array(self.indices)[:, ~corner_voxels])
    self.prescribed = self.prescribed[~corner_voxels]
    return

jax_lab.core.boundary_conditions.Regularized

Bases: ZouHe

Regularized boundary condition for a lattice Boltzmann method simulation.

This class implements the regularized boundary condition, which is a non-equilibrium bounce-back boundary condition with additional regularization. It can be used to set inflow and outflow boundary conditions with prescribed pressure or velocity.

Attributes

name (str): The name of the boundary condition. For this class, it is “Regularized”.

Qi (numpy.ndarray): The Qi tensor, which is used in the regularization of the distribution functions.

References

Latt, J. (2007). Hydrodynamic limit of lattice Boltzmann equations. PhD thesis, University of Geneva.

Latt, J., Chopard, B., Malaspinas, O., Deville, M., & Michler, A. (2008). Straight velocity boundaries in the

lattice Boltzmann method. Physical Review E, 77(5), 056703. doi:10.1103/PhysRevE.77.056703

Source code in jax_lab/core/boundary_conditions.py
class Regularized(ZouHe):
    """
    Regularized boundary condition for a lattice Boltzmann method simulation.

    This class implements the regularized boundary condition, which is a non-equilibrium bounce-back boundary condition
    with additional regularization. It can be used to set inflow and outflow boundary conditions with prescribed pressure
    or velocity.

    Attributes
    ----------
    name (str): The name of the boundary condition. For this class, it is "Regularized".

    Qi (numpy.ndarray): The Qi tensor, which is used in the regularization of the distribution functions.

    References
    ----------
    Latt, J. (2007). Hydrodynamic limit of lattice Boltzmann equations. PhD thesis, University of Geneva.

    Latt, J., Chopard, B., Malaspinas, O., Deville, M., & Michler, A. (2008). Straight velocity boundaries in the

    lattice Boltzmann method. Physical Review E, 77(5), 056703. doi:10.1103/PhysRevE.77.056703
    """

    def __init__(self, indices, grid_info, precision_policy, ttype, prescribed):
        super().__init__(indices, grid_info, precision_policy, ttype, prescribed)
        self.name = "Regularized"
        # TODO for Hesam: check to understand why corner cases cause instability here.
        # self.needs_extra_configuration = False
        self.construct_symmetric_lattice_moment()

    def construct_symmetric_lattice_moment(self):
        """
        Construct the symmetric lattice moment Qi.

        The Qi tensor is used in the regularization of the distribution functions. It is defined as Qi = cc - cs^2*I,
        where cc is the tensor of lattice velocities, cs is the speed of sound, and I is the identity tensor.
        """
        self.Qi = _construct_symmetric_lattice_moment(self.lattice.cc, self.dim)
        return

    @partial(jit, static_argnums=(0,), inline=True)
    def regularize_fpop(self, fpop, feq):
        """
        Regularizes the distribution functions by adding non-equilibrium contributions based on second moments of fpop.

        Parameters
        ----------
        fpop (jax.numpy.ndarray): The distribution functions.

        feq (jax.numpy.ndarray): The equilibrium distribution functions.

        Returns
        -------
        (jax.numpy.ndarray): The regularized distribution functions.
        """

        # Compute momentum flux of off-equilibrium populations for regularization: Pi^1 = Pi^{neq}
        f_neq = fpop - feq
        PiNeq = self.momentum_flux(f_neq)
        # PiNeq = self.momentum_flux(fpop) - self.momentum_flux(feq)

        # Compute double dot product Qi:Pi1
        # QiPi1 = np.zeros_like(fpop)
        # Pi1 = PiNeq
        # QiPi1 = jnp.dot(Qi, Pi1)
        QiPi1 = jnp.dot(PiNeq, self.Qi)

        # assign all populations based on eq 45 of Latt et al (2008)
        # fneq ~ f^1
        fpop1 = 9.0 / 2.0 * self.lattice.w[None, :] * QiPi1
        fpop_regularized = feq + fpop1

        # feq/fpop1 are at compute precision while fpop (from bounceback_nonequilibrium, i.e. fout) is at output
        # precision (e.g. "f32/f16"): cast explicitly instead of relying on an implicit narrowing cast in the
        # caller's .set(), which JAX warns will become an error in a future release.
        return fpop_regularized.astype(fpop.dtype)

    @partial(jit, static_argnums=(0,))
    def apply(self, fout, _):
        """
        Applies the regularized boundary condition.

        Parameters
        ----------
        fout (jax.numpy.ndarray): The output distribution functions.

        _ (jax.numpy.ndarray): The input distribution functions. This is not used in this method.

        Returns
        -------
        (jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.

        Notes
        -----
        This method applies the regularized boundary condition by first computing the equilibrium distribution functions based
        on the prescribed values and the type of boundary condition, then setting the unknown distribution functions
        based on the non-equilibrium bounce-back method, and finally regularizing the distribution functions.
        """

        # compute the equilibrium based on prescribed values and the type of BC
        feq = self.calculate_equilibrium(fout)

        # set the unknown f populations based on the non-equilibrium bounce-back method
        fbd = self.bounceback_nonequilibrium(fout, feq)

        # Regularize the boundary fpop
        fbd = self.regularize_fpop(fbd, feq)
        return fbd

apply

apply(fout, _)

Applies the regularized boundary condition.

Parameters

fout (jax.numpy.ndarray): The output distribution functions.

_ (jax.numpy.ndarray): The input distribution functions. This is not used in this method.

Returns

(jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.

Notes

This method applies the regularized boundary condition by first computing the equilibrium distribution functions based on the prescribed values and the type of boundary condition, then setting the unknown distribution functions based on the non-equilibrium bounce-back method, and finally regularizing the distribution functions.

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0,))
def apply(self, fout, _):
    """
    Applies the regularized boundary condition.

    Parameters
    ----------
    fout (jax.numpy.ndarray): The output distribution functions.

    _ (jax.numpy.ndarray): The input distribution functions. This is not used in this method.

    Returns
    -------
    (jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.

    Notes
    -----
    This method applies the regularized boundary condition by first computing the equilibrium distribution functions based
    on the prescribed values and the type of boundary condition, then setting the unknown distribution functions
    based on the non-equilibrium bounce-back method, and finally regularizing the distribution functions.
    """

    # compute the equilibrium based on prescribed values and the type of BC
    feq = self.calculate_equilibrium(fout)

    # set the unknown f populations based on the non-equilibrium bounce-back method
    fbd = self.bounceback_nonequilibrium(fout, feq)

    # Regularize the boundary fpop
    fbd = self.regularize_fpop(fbd, feq)
    return fbd

construct_symmetric_lattice_moment

construct_symmetric_lattice_moment()

Construct the symmetric lattice moment Qi.

The Qi tensor is used in the regularization of the distribution functions. It is defined as Qi = cc - cs^2*I, where cc is the tensor of lattice velocities, cs is the speed of sound, and I is the identity tensor.

Source code in jax_lab/core/boundary_conditions.py
def construct_symmetric_lattice_moment(self):
    """
    Construct the symmetric lattice moment Qi.

    The Qi tensor is used in the regularization of the distribution functions. It is defined as Qi = cc - cs^2*I,
    where cc is the tensor of lattice velocities, cs is the speed of sound, and I is the identity tensor.
    """
    self.Qi = _construct_symmetric_lattice_moment(self.lattice.cc, self.dim)
    return

regularize_fpop

regularize_fpop(fpop, feq)

Regularizes the distribution functions by adding non-equilibrium contributions based on second moments of fpop.

Parameters

fpop (jax.numpy.ndarray): The distribution functions.

feq (jax.numpy.ndarray): The equilibrium distribution functions.

Returns

(jax.numpy.ndarray): The regularized distribution functions.

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0,), inline=True)
def regularize_fpop(self, fpop, feq):
    """
    Regularizes the distribution functions by adding non-equilibrium contributions based on second moments of fpop.

    Parameters
    ----------
    fpop (jax.numpy.ndarray): The distribution functions.

    feq (jax.numpy.ndarray): The equilibrium distribution functions.

    Returns
    -------
    (jax.numpy.ndarray): The regularized distribution functions.
    """

    # Compute momentum flux of off-equilibrium populations for regularization: Pi^1 = Pi^{neq}
    f_neq = fpop - feq
    PiNeq = self.momentum_flux(f_neq)
    # PiNeq = self.momentum_flux(fpop) - self.momentum_flux(feq)

    # Compute double dot product Qi:Pi1
    # QiPi1 = np.zeros_like(fpop)
    # Pi1 = PiNeq
    # QiPi1 = jnp.dot(Qi, Pi1)
    QiPi1 = jnp.dot(PiNeq, self.Qi)

    # assign all populations based on eq 45 of Latt et al (2008)
    # fneq ~ f^1
    fpop1 = 9.0 / 2.0 * self.lattice.w[None, :] * QiPi1
    fpop_regularized = feq + fpop1

    # feq/fpop1 are at compute precision while fpop (from bounceback_nonequilibrium, i.e. fout) is at output
    # precision (e.g. "f32/f16"): cast explicitly instead of relying on an implicit narrowing cast in the
    # caller's .set(), which JAX warns will become an error in a future release.
    return fpop_regularized.astype(fpop.dtype)

jax_lab.core.boundary_conditions.ExtrapolationOutflow

Bases: BoundaryCondition

Extrapolation outflow boundary condition for a lattice Boltzmann method simulation.

This class implements the extrapolation outflow boundary condition, which is a type of outflow boundary condition that uses extrapolation to avoid strong wave reflections.

Attributes

name (str): The name of the boundary condition. For this class, it is “ExtrapolationOutflow”.

sound_speed (float): The speed of sound in the simulation.

References

Geier, M., Schönherr, M., Pasquali, A., & Krafczyk, M. (2015). The cumulant lattice Boltzmann equation in three dimensions: Theory and validation. Computers & Mathematics with Applications, 70(4), 507-547. doi:10.1016/j.camwa.2015.05.001.

Source code in jax_lab/core/boundary_conditions.py
class ExtrapolationOutflow(BoundaryCondition):
    """
    Extrapolation outflow boundary condition for a lattice Boltzmann method simulation.

    This class implements the extrapolation outflow boundary condition, which is a type of outflow boundary condition
    that uses extrapolation to avoid strong wave reflections.

    Attributes
    ----------
    name (str): The name of the boundary condition. For this class, it is "ExtrapolationOutflow".

    sound_speed (float): The speed of sound in the simulation.

    References
    ----------
    Geier, M., Schönherr, M., Pasquali, A., & Krafczyk, M. (2015). The cumulant lattice Boltzmann equation in three
    dimensions: Theory and validation. Computers & Mathematics with Applications, 70(4), 507-547.
    doi:10.1016/j.camwa.2015.05.001.
    """

    def __init__(self, indices, grid_info, precision_policy):
        super().__init__(indices, grid_info, precision_policy)
        self.name = "ExtrapolationOutflow"
        self.needs_extra_configuration = True
        self.sound_speed = 1.0 / jnp.sqrt(3.0)

    def configure(self, boundary_mask):
        """
        Configure one inward fluid neighbour for every boundary node.

        Parameters
        ----------
        boundary_mask (np.ndarray): The grid mask for the boundary voxels.
        """
        idx = np.array(self.indices).T
        has_fluid_neighbour = ~boundary_mask[:, self.lattice.opp_indices]
        if not np.all(np.any(has_fluid_neighbour, axis=1)):
            raise ValueError("ExtrapolationOutflow requires at least one fluid neighbour per boundary node.")

        # Preserve one-to-one correspondence between boundary and neighbour nodes.
        direction = np.argmax(has_fluid_neighbour, axis=1)
        indices_nbr = idx + self.lattice.c[:, direction].T
        self.indices_nbr = tuple(indices_nbr.T)

        return

    @partial(jit, static_argnums=(0, 3), inline=True)
    def prepare_populations(self, fout, fin, implementation_step):
        """
        Prepares the distribution functions for the boundary condition.

        Parameters
        ----------
        fout (jax.numpy.ndarray): The incoming distribution functions.

        fin (jax.numpy.ndarray): The outgoing distribution functions.

        implementation_step (str): The step in the lattice Boltzmann method algorithm at which the preparation is
            applied.

        Returns
        -------
        (jax.numpy.ndarray): The prepared distribution functions.

        Notes
        -----
        Because this function is called "PostCollision", f_poststreaming refers to previous time step or t-1
        """
        f_postcollision = fout
        f_poststreaming = fin
        if implementation_step == "PostStreaming":
            return f_postcollision
        nbd = len(self.indices[0])
        bindex = np.arange(nbd)[:, None]
        fps_bdr = f_poststreaming[self.indices]
        fps_nbr = f_poststreaming[self.indices_nbr]
        fpc_bdr = f_postcollision[self.indices]
        fpop = fps_bdr[bindex, self.imissing]
        fpop_neighbour = fps_nbr[bindex, self.imissing]
        fpop_extrapolated = self.sound_speed * fpop_neighbour + (1.0 - self.sound_speed) * fpop

        # Use the iknown directions of f_postcollision that leave the domain during streaming to store the BC data
        fpc_bdr = fpc_bdr.at[bindex, self.iknown].set(fpop_extrapolated)
        f_postcollision = f_postcollision.at[self.indices].set(fpc_bdr)
        return f_postcollision

    @partial(jit, static_argnums=(0,))
    def apply(self, fout, fin):
        """
        Applies the extrapolation outflow boundary condition.

        Parameters
        ----------
        fout (jax.numpy.ndarray): The output distribution functions.

        fin (jax.numpy.ndarray): The input distribution functions.

        Returns
        -------
        (jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.
        """
        nbd = len(self.indices[0])
        bindex = np.arange(nbd)[:, None]
        fbd = fout[self.indices]
        fbd = fbd.at[bindex, self.imissing].set(fin[self.indices][bindex, self.iknown])
        return fbd

apply

apply(fout, fin)

Applies the extrapolation outflow boundary condition.

Parameters

fout (jax.numpy.ndarray): The output distribution functions.

fin (jax.numpy.ndarray): The input distribution functions.

Returns

(jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0,))
def apply(self, fout, fin):
    """
    Applies the extrapolation outflow boundary condition.

    Parameters
    ----------
    fout (jax.numpy.ndarray): The output distribution functions.

    fin (jax.numpy.ndarray): The input distribution functions.

    Returns
    -------
    (jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.
    """
    nbd = len(self.indices[0])
    bindex = np.arange(nbd)[:, None]
    fbd = fout[self.indices]
    fbd = fbd.at[bindex, self.imissing].set(fin[self.indices][bindex, self.iknown])
    return fbd

configure

configure(boundary_mask)

Configure one inward fluid neighbour for every boundary node.

Parameters

boundary_mask (np.ndarray): The grid mask for the boundary voxels.

Source code in jax_lab/core/boundary_conditions.py
def configure(self, boundary_mask):
    """
    Configure one inward fluid neighbour for every boundary node.

    Parameters
    ----------
    boundary_mask (np.ndarray): The grid mask for the boundary voxels.
    """
    idx = np.array(self.indices).T
    has_fluid_neighbour = ~boundary_mask[:, self.lattice.opp_indices]
    if not np.all(np.any(has_fluid_neighbour, axis=1)):
        raise ValueError("ExtrapolationOutflow requires at least one fluid neighbour per boundary node.")

    # Preserve one-to-one correspondence between boundary and neighbour nodes.
    direction = np.argmax(has_fluid_neighbour, axis=1)
    indices_nbr = idx + self.lattice.c[:, direction].T
    self.indices_nbr = tuple(indices_nbr.T)

    return

prepare_populations

prepare_populations(fout, fin, implementation_step)

Prepares the distribution functions for the boundary condition.

Parameters

fout (jax.numpy.ndarray): The incoming distribution functions.

fin (jax.numpy.ndarray): The outgoing distribution functions.

implementation_step (str): The step in the lattice Boltzmann method algorithm at which the preparation is applied.

Returns

(jax.numpy.ndarray): The prepared distribution functions.

Notes

Because this function is called “PostCollision”, f_poststreaming refers to previous time step or t-1

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0, 3), inline=True)
def prepare_populations(self, fout, fin, implementation_step):
    """
    Prepares the distribution functions for the boundary condition.

    Parameters
    ----------
    fout (jax.numpy.ndarray): The incoming distribution functions.

    fin (jax.numpy.ndarray): The outgoing distribution functions.

    implementation_step (str): The step in the lattice Boltzmann method algorithm at which the preparation is
        applied.

    Returns
    -------
    (jax.numpy.ndarray): The prepared distribution functions.

    Notes
    -----
    Because this function is called "PostCollision", f_poststreaming refers to previous time step or t-1
    """
    f_postcollision = fout
    f_poststreaming = fin
    if implementation_step == "PostStreaming":
        return f_postcollision
    nbd = len(self.indices[0])
    bindex = np.arange(nbd)[:, None]
    fps_bdr = f_poststreaming[self.indices]
    fps_nbr = f_poststreaming[self.indices_nbr]
    fpc_bdr = f_postcollision[self.indices]
    fpop = fps_bdr[bindex, self.imissing]
    fpop_neighbour = fps_nbr[bindex, self.imissing]
    fpop_extrapolated = self.sound_speed * fpop_neighbour + (1.0 - self.sound_speed) * fpop

    # Use the iknown directions of f_postcollision that leave the domain during streaming to store the BC data
    fpc_bdr = fpc_bdr.at[bindex, self.iknown].set(fpop_extrapolated)
    f_postcollision = f_postcollision.at[self.indices].set(fpc_bdr)
    return f_postcollision

jax_lab.core.boundary_conditions.InterpolatedBounceBackBouzidi

Bases: BounceBackHalfway

A local single-node version of the interpolated bounce-back boundary condition due to Bouzidi for a lattice Boltzmann method simulation.

This class implements a interpolated bounce-back boundary condition. The boundary condition is applied after the streaming step.

Attributes

name (str): The name of the boundary condition. For this class, it is “InterpolatedBounceBackBouzidi”. implicit_distances (array-like): An array of shape (nx,ny,nz) indicating the signed-distance field from the solid walls weights (array-like): An array of shape (number_of_bc_cells, q) initialized as None and constructed using implicit_distances array during runtime. These “weights” are associated with the fractional distance of fluid cell to the boundary position defined as: weights(dir_i) = |x_fluid - x_boundary(dir_i)| / |x_fluid - x_solid(dir_i)|.

theta (jax.numpy.ndarray; Default: None): Contact angle, applied for multiphase flows and only set for wall boundary conditions.

phi (jax.numpy.ndarray; Default: None): Contact angle parameter phi, applied for multiphase flows and only set for wall boundary conditions.

delta_rho (jax.numpy.ndarray; Default: None): Contact angle parameter delta_rho, applied for multiphase flows and only set for wall boundary conditions.

Source code in jax_lab/core/boundary_conditions.py
class InterpolatedBounceBackBouzidi(BounceBackHalfway):
    """
    A local single-node version of the interpolated bounce-back boundary condition due to Bouzidi for a lattice
    Boltzmann method simulation.

    This class implements a interpolated bounce-back boundary condition. The boundary condition is applied after
    the streaming step.

    Attributes
    ----------
    name (str): The name of the boundary condition. For this class, it is "InterpolatedBounceBackBouzidi".
    implicit_distances (array-like): An array of shape (nx,ny,nz) indicating the signed-distance field from the solid walls
    weights (array-like): An array of shape (number_of_bc_cells, q) initialized as None and constructed using implicit_distances array
    during runtime. These "weights" are associated with the fractional distance of fluid cell to the boundary
    position defined as: weights(dir_i) = |x_fluid - x_boundary(dir_i)| / |x_fluid - x_solid(dir_i)|.

    theta (jax.numpy.ndarray; Default: None): Contact angle, applied for multiphase flows and only set for wall boundary conditions.

    phi (jax.numpy.ndarray; Default: None): Contact angle parameter phi, applied for multiphase flows and only set for wall boundary conditions.

    delta_rho (jax.numpy.ndarray; Default: None): Contact angle parameter delta_rho, applied for multiphase flows and only set for wall boundary conditions.
    """

    def __init__(self, indices, implicit_distances, grid_info, precision_policy, vel=None, theta=None, phi=None, delta_rho=None):
        super().__init__(indices, grid_info, precision_policy, vel=vel)
        self.name = "InterpolatedBounceBackBouzidi"
        self.implicit_distances = implicit_distances
        self.weights = None
        self.theta = theta
        self.phi = phi
        self.delta_rho = delta_rho

    def set_proximity_ratio(self):
        """
        Creates the interpolation data needed for the boundary condition.

        Returns
        -------
        None. The function updates the object's weights attribute in place.
        """
        epsilon = 1e-12
        nbd = len(self.indices[0])
        idx = np.array(self.indices).T
        bindex = np.arange(nbd)[:, None]
        weights = np.full((idx.shape[0], self.lattice.q), 0.5)
        c = np.array(self.lattice.c)
        sdf_f = self.implicit_distances[self.indices]
        for q in range(1, self.lattice.q):
            solid_indices = idx + c[:, q]
            solid_indices_tuple = tuple(map(tuple, solid_indices.T))
            sdf_s = self.implicit_distances[solid_indices_tuple]
            weights[:, q] = sdf_f / (sdf_f - sdf_s + epsilon)
        self.weights = weights[bindex, self.iknown]
        return

    @staticmethod
    def interpolate_missing(fbd, fin_bd, fout_bd, imissing, iknown, weights):
        """
        Core Bouzidi interpolated bounce-back math. Pure function of its arguments (no BC-instance state), so it
        is reusable both by apply() (global indices) and by a local (per-shard) bounce-back kernel operating on
        a gathered block with the same (n, q) shape.

        Parameters
        ----------
        fbd (jax.numpy.ndarray): Post-streaming distribution functions at the boundary nodes, shape (n, q).

        fin_bd (jax.numpy.ndarray): Post-collision distribution functions at the boundary nodes, shape (n, q).

        fout_bd (jax.numpy.ndarray): Post-streaming distribution functions at the boundary nodes, shape (n, q).

        imissing (array-like): Missing-direction indices, shape (n, q).

        iknown (array-like): Known (opposite) direction indices, shape (n, q).

        weights (array-like): Proximity-ratio interpolation weights, shape (n, q).

        Returns
        -------
        (jax.numpy.ndarray): fbd with each missing direction set from the interpolated value.
        """
        bindex = jnp.arange(fbd.shape[0])[:, None]
        f_postcollision_iknown = fin_bd[bindex, iknown]
        f_postcollision_imissing = fin_bd[bindex, imissing]
        f_poststreaming_iknown = fout_bd[bindex, iknown]

        # if weights<0.5
        fs_near = 2.0 * weights * f_postcollision_iknown + (1.0 - 2.0 * weights) * f_poststreaming_iknown

        # if weights>=0.5
        fs_far = 1.0 / (2.0 * weights) * f_postcollision_iknown + (2.0 * weights - 1.0) / (2.0 * weights) * f_postcollision_imissing

        # combine near and far contributions
        fmissing = jnp.where(weights < 0.5, fs_near, fs_far)
        return fbd.at[bindex, imissing].set(fmissing)

    @partial(jit, static_argnums=(0,))
    def apply(self, fout, fin):
        """
        Applies the halfway bounce-back boundary condition.

        Parameters
        ----------
        fout (jax.numpy.ndarray): The output distribution functions.

        fin (jax.numpy.ndarray): The input distribution functions.

        Returns
        -------
        (jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.
        """
        if self.weights is None:
            self.set_proximity_ratio()
        fbd = self.interpolate_missing(fout[self.indices], fin[self.indices], fout[self.indices], self.imissing, self.iknown, self.weights)

        if self.vel is not None:
            bindex = np.arange(len(self.indices[0]))[:, None]
            fbd = self.impose_boundary_vel(fbd, bindex)
        return fbd

apply

apply(fout, fin)

Applies the halfway bounce-back boundary condition.

Parameters

fout (jax.numpy.ndarray): The output distribution functions.

fin (jax.numpy.ndarray): The input distribution functions.

Returns

(jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0,))
def apply(self, fout, fin):
    """
    Applies the halfway bounce-back boundary condition.

    Parameters
    ----------
    fout (jax.numpy.ndarray): The output distribution functions.

    fin (jax.numpy.ndarray): The input distribution functions.

    Returns
    -------
    (jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.
    """
    if self.weights is None:
        self.set_proximity_ratio()
    fbd = self.interpolate_missing(fout[self.indices], fin[self.indices], fout[self.indices], self.imissing, self.iknown, self.weights)

    if self.vel is not None:
        bindex = np.arange(len(self.indices[0]))[:, None]
        fbd = self.impose_boundary_vel(fbd, bindex)
    return fbd

interpolate_missing staticmethod

interpolate_missing(fbd, fin_bd, fout_bd, imissing, iknown, weights)

Core Bouzidi interpolated bounce-back math. Pure function of its arguments (no BC-instance state), so it is reusable both by apply() (global indices) and by a local (per-shard) bounce-back kernel operating on a gathered block with the same (n, q) shape.

Parameters

fbd (jax.numpy.ndarray): Post-streaming distribution functions at the boundary nodes, shape (n, q).

fin_bd (jax.numpy.ndarray): Post-collision distribution functions at the boundary nodes, shape (n, q).

fout_bd (jax.numpy.ndarray): Post-streaming distribution functions at the boundary nodes, shape (n, q).

imissing (array-like): Missing-direction indices, shape (n, q).

iknown (array-like): Known (opposite) direction indices, shape (n, q).

weights (array-like): Proximity-ratio interpolation weights, shape (n, q).

Returns

(jax.numpy.ndarray): fbd with each missing direction set from the interpolated value.

Source code in jax_lab/core/boundary_conditions.py
@staticmethod
def interpolate_missing(fbd, fin_bd, fout_bd, imissing, iknown, weights):
    """
    Core Bouzidi interpolated bounce-back math. Pure function of its arguments (no BC-instance state), so it
    is reusable both by apply() (global indices) and by a local (per-shard) bounce-back kernel operating on
    a gathered block with the same (n, q) shape.

    Parameters
    ----------
    fbd (jax.numpy.ndarray): Post-streaming distribution functions at the boundary nodes, shape (n, q).

    fin_bd (jax.numpy.ndarray): Post-collision distribution functions at the boundary nodes, shape (n, q).

    fout_bd (jax.numpy.ndarray): Post-streaming distribution functions at the boundary nodes, shape (n, q).

    imissing (array-like): Missing-direction indices, shape (n, q).

    iknown (array-like): Known (opposite) direction indices, shape (n, q).

    weights (array-like): Proximity-ratio interpolation weights, shape (n, q).

    Returns
    -------
    (jax.numpy.ndarray): fbd with each missing direction set from the interpolated value.
    """
    bindex = jnp.arange(fbd.shape[0])[:, None]
    f_postcollision_iknown = fin_bd[bindex, iknown]
    f_postcollision_imissing = fin_bd[bindex, imissing]
    f_poststreaming_iknown = fout_bd[bindex, iknown]

    # if weights<0.5
    fs_near = 2.0 * weights * f_postcollision_iknown + (1.0 - 2.0 * weights) * f_poststreaming_iknown

    # if weights>=0.5
    fs_far = 1.0 / (2.0 * weights) * f_postcollision_iknown + (2.0 * weights - 1.0) / (2.0 * weights) * f_postcollision_imissing

    # combine near and far contributions
    fmissing = jnp.where(weights < 0.5, fs_near, fs_far)
    return fbd.at[bindex, imissing].set(fmissing)

set_proximity_ratio

set_proximity_ratio()

Creates the interpolation data needed for the boundary condition.

Returns

None. The function updates the object’s weights attribute in place.

Source code in jax_lab/core/boundary_conditions.py
def set_proximity_ratio(self):
    """
    Creates the interpolation data needed for the boundary condition.

    Returns
    -------
    None. The function updates the object's weights attribute in place.
    """
    epsilon = 1e-12
    nbd = len(self.indices[0])
    idx = np.array(self.indices).T
    bindex = np.arange(nbd)[:, None]
    weights = np.full((idx.shape[0], self.lattice.q), 0.5)
    c = np.array(self.lattice.c)
    sdf_f = self.implicit_distances[self.indices]
    for q in range(1, self.lattice.q):
        solid_indices = idx + c[:, q]
        solid_indices_tuple = tuple(map(tuple, solid_indices.T))
        sdf_s = self.implicit_distances[solid_indices_tuple]
        weights[:, q] = sdf_f / (sdf_f - sdf_s + epsilon)
    self.weights = weights[bindex, self.iknown]
    return

jax_lab.core.boundary_conditions.InterpolatedBounceBackDifferentiable

Bases: InterpolatedBounceBackBouzidi

A differentiable variant of the “InterpolatedBounceBackBouzidi” BC scheme. This BC is now differentiable at self.weight = 0.5 unlike the original Bouzidi scheme which switches between 2 equations at weight=0.5. Refer to [1] (their Appendix E) for more information.

References

[1] Geier, M., Schönherr, M., Pasquali, A., & Krafczyk, M. (2015). The cumulant lattice Boltzmann equation in three dimensions: Theory and validation. Computers & Mathematics with Applications, 70(4), 507-547. doi:10.1016/j.camwa.2015.05.001.

This class implements a interpolated bounce-back boundary condition. The boundary condition is applied after the streaming step.

Attributes

name (str): The name of the boundary condition. For this class, it is “InterpolatedBounceBackDifferentiable”.

Source code in jax_lab/core/boundary_conditions.py
class InterpolatedBounceBackDifferentiable(InterpolatedBounceBackBouzidi):
    """
    A differentiable variant of the "InterpolatedBounceBackBouzidi" BC scheme. This BC is now differentiable at
    self.weight = 0.5 unlike the original Bouzidi scheme which switches between 2 equations at weight=0.5. Refer to
    [1] (their Appendix E) for more information.

    References
    ----------
    [1] Geier, M., Schönherr, M., Pasquali, A., & Krafczyk, M. (2015). The cumulant lattice Boltzmann equation in three
    dimensions: Theory and validation. Computers & Mathematics with Applications, 70(4), 507-547.
    doi:10.1016/j.camwa.2015.05.001.


    This class implements a interpolated bounce-back boundary condition. The boundary condition is applied after
    the streaming step.

    Attributes
    ----------
    name (str): The name of the boundary condition. For this class, it is "InterpolatedBounceBackDifferentiable".
    """

    def __init__(self, indices, implicit_distances, grid_info, precision_policy, vel=None):
        super().__init__(indices, implicit_distances, grid_info, precision_policy, vel=vel)
        self.name = "InterpolatedBounceBackDifferentiable"

    @staticmethod
    def interpolate_missing(fbd, fin_bd, fout_bd, imissing, iknown, weights):
        """
        Core differentiable interpolated bounce-back math. Pure function of its arguments (no BC-instance
        state), so it is reusable both by apply() (global indices) and by a local (per-shard) bounce-back
        kernel operating on a gathered block with the same (n, q) shape.

        Parameters
        ----------
        fbd (jax.numpy.ndarray): Post-streaming distribution functions at the boundary nodes, shape (n, q).

        fin_bd (jax.numpy.ndarray): Post-collision distribution functions at the boundary nodes, shape (n, q).

        fout_bd (jax.numpy.ndarray): Post-streaming distribution functions at the boundary nodes, shape (n, q).

        imissing (array-like): Missing-direction indices, shape (n, q).

        iknown (array-like): Known (opposite) direction indices, shape (n, q).

        weights (array-like): Proximity-ratio interpolation weights, shape (n, q).

        Returns
        -------
        (jax.numpy.ndarray): fbd with each missing direction set from the interpolated value.
        """
        bindex = jnp.arange(fbd.shape[0])[:, None]
        f_postcollision_iknown = fin_bd[bindex, iknown]
        f_postcollision_imissing = fin_bd[bindex, imissing]
        f_poststreaming_iknown = fout_bd[bindex, iknown]
        fmissing = ((1.0 - weights) * f_poststreaming_iknown + weights * (f_postcollision_imissing + f_postcollision_iknown)) / (1.0 + weights)
        return fbd.at[bindex, imissing].set(fmissing)

    @partial(jit, static_argnums=(0,))
    def apply(self, fout, fin):
        """
        Applies the halfway bounce-back boundary condition.

        Parameters
        ----------
        fout (jax.numpy.ndarray): The output distribution functions.

        fin (jax.numpy.ndarray): The input distribution functions.

        Returns
        -------
        jax.numpy.ndarray
            The modified output distribution functions after applying the boundary condition.
        """
        if self.weights is None:
            self.set_proximity_ratio()
        fbd = self.interpolate_missing(fout[self.indices], fin[self.indices], fout[self.indices], self.imissing, self.iknown, self.weights)

        if self.vel is not None:
            bindex = np.arange(len(self.indices[0]))[:, None]
            fbd = self.impose_boundary_vel(fbd, bindex)
        return fbd

apply

apply(fout, fin)

Applies the halfway bounce-back boundary condition.

Parameters

fout (jax.numpy.ndarray): The output distribution functions.

fin (jax.numpy.ndarray): The input distribution functions.

Returns

jax.numpy.ndarray The modified output distribution functions after applying the boundary condition.

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0,))
def apply(self, fout, fin):
    """
    Applies the halfway bounce-back boundary condition.

    Parameters
    ----------
    fout (jax.numpy.ndarray): The output distribution functions.

    fin (jax.numpy.ndarray): The input distribution functions.

    Returns
    -------
    jax.numpy.ndarray
        The modified output distribution functions after applying the boundary condition.
    """
    if self.weights is None:
        self.set_proximity_ratio()
    fbd = self.interpolate_missing(fout[self.indices], fin[self.indices], fout[self.indices], self.imissing, self.iknown, self.weights)

    if self.vel is not None:
        bindex = np.arange(len(self.indices[0]))[:, None]
        fbd = self.impose_boundary_vel(fbd, bindex)
    return fbd

interpolate_missing staticmethod

interpolate_missing(fbd, fin_bd, fout_bd, imissing, iknown, weights)

Core differentiable interpolated bounce-back math. Pure function of its arguments (no BC-instance state), so it is reusable both by apply() (global indices) and by a local (per-shard) bounce-back kernel operating on a gathered block with the same (n, q) shape.

Parameters

fbd (jax.numpy.ndarray): Post-streaming distribution functions at the boundary nodes, shape (n, q).

fin_bd (jax.numpy.ndarray): Post-collision distribution functions at the boundary nodes, shape (n, q).

fout_bd (jax.numpy.ndarray): Post-streaming distribution functions at the boundary nodes, shape (n, q).

imissing (array-like): Missing-direction indices, shape (n, q).

iknown (array-like): Known (opposite) direction indices, shape (n, q).

weights (array-like): Proximity-ratio interpolation weights, shape (n, q).

Returns

(jax.numpy.ndarray): fbd with each missing direction set from the interpolated value.

Source code in jax_lab/core/boundary_conditions.py
@staticmethod
def interpolate_missing(fbd, fin_bd, fout_bd, imissing, iknown, weights):
    """
    Core differentiable interpolated bounce-back math. Pure function of its arguments (no BC-instance
    state), so it is reusable both by apply() (global indices) and by a local (per-shard) bounce-back
    kernel operating on a gathered block with the same (n, q) shape.

    Parameters
    ----------
    fbd (jax.numpy.ndarray): Post-streaming distribution functions at the boundary nodes, shape (n, q).

    fin_bd (jax.numpy.ndarray): Post-collision distribution functions at the boundary nodes, shape (n, q).

    fout_bd (jax.numpy.ndarray): Post-streaming distribution functions at the boundary nodes, shape (n, q).

    imissing (array-like): Missing-direction indices, shape (n, q).

    iknown (array-like): Known (opposite) direction indices, shape (n, q).

    weights (array-like): Proximity-ratio interpolation weights, shape (n, q).

    Returns
    -------
    (jax.numpy.ndarray): fbd with each missing direction set from the interpolated value.
    """
    bindex = jnp.arange(fbd.shape[0])[:, None]
    f_postcollision_iknown = fin_bd[bindex, iknown]
    f_postcollision_imissing = fin_bd[bindex, imissing]
    f_poststreaming_iknown = fout_bd[bindex, iknown]
    fmissing = ((1.0 - weights) * f_poststreaming_iknown + weights * (f_postcollision_imissing + f_postcollision_iknown)) / (1.0 + weights)
    return fbd.at[bindex, imissing].set(fmissing)

jax_lab.core.boundary_conditions.ExtrapolationOutflowMultiphase

Bases: BoundaryCondition

Extrapolation boundary condition for multiphase flows.

Attributes

name (str): The name of the boundary condition. For this class, it is “NonEquilibriumExtrapolation”.

References

  1. Zhao-Li, G., Chu-Guang, Z. & Bao-Chang, S. Non-equilibrium extrapolation method for velocity and pressure boundary conditions in the lattice Boltzmann method. Chinese Phys. 11, 366-374 (2002).
Source code in jax_lab/core/boundary_conditions.py
class ExtrapolationOutflowMultiphase(BoundaryCondition):
    """
    Extrapolation boundary condition for multiphase flows.

    Attributes
    ----------
    name (str): The name of the boundary condition. For this class, it is "NonEquilibriumExtrapolation".

    References
    ----------
    1. Zhao-Li, G., Chu-Guang, Z. & Bao-Chang, S. Non-equilibrium extrapolation method for velocity and pressure boundary conditions in the lattice
    Boltzmann method. Chinese Phys. 11, 366-374 (2002).
    """

    def __init__(self, indices, grid_info, precision_policy):
        super().__init__(indices, grid_info, precision_policy)
        self.name = "ExtrapolationOutflowMultiphase"
        self.needs_extra_configuration = False
        self.neighbors_found = False

    def find_neighbors(self):
        ind = np.array(self.indices).T - self.normals
        self.indices_nbr = tuple(ind.T)
        self.indices_next_nbr = tuple((np.array(self.indices_nbr).T - self.normals).T)
        nbd = len(self.indices[0])
        self.bindex = np.arange(nbd)[:, None]

    @partial(jit, static_argnums=(0,))
    def apply(self, fout, _):
        """
        Applies the non-equilibrium extrapolation boundary condition.

        Parameters
        ----------
        fout (jax.numpy.ndarray): The output distribution functions.

        _ (jax.numpy.ndarray): The input distribution functions, not used in this function

        Returns
        -------
        jax.numpy.ndarray
            The modified output distribution functions after applying the boundary condition.
        """
        if not self.neighbors_found:
            self.find_neighbors()
            self.neighbors_found = True

        fbd = fout[self.indices]
        f_nbr = fout[self.indices_nbr]
        f_next_nbr = fout[self.indices_next_nbr]
        # fbd = fbd.at[self.bindex, ...].set(2 * f_nbr[self.bindex, ...] - f_next_nbr[self.bindex, ...])
        fbd = fbd.at[self.bindex, self.imissing].set(2 * f_nbr[self.bindex, self.imissing] - f_next_nbr[self.bindex, self.imissing])
        return fbd

apply

apply(fout, _)

Applies the non-equilibrium extrapolation boundary condition.

Parameters

fout (jax.numpy.ndarray): The output distribution functions.

_ (jax.numpy.ndarray): The input distribution functions, not used in this function

Returns

jax.numpy.ndarray The modified output distribution functions after applying the boundary condition.

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0,))
def apply(self, fout, _):
    """
    Applies the non-equilibrium extrapolation boundary condition.

    Parameters
    ----------
    fout (jax.numpy.ndarray): The output distribution functions.

    _ (jax.numpy.ndarray): The input distribution functions, not used in this function

    Returns
    -------
    jax.numpy.ndarray
        The modified output distribution functions after applying the boundary condition.
    """
    if not self.neighbors_found:
        self.find_neighbors()
        self.neighbors_found = True

    fbd = fout[self.indices]
    f_nbr = fout[self.indices_nbr]
    f_next_nbr = fout[self.indices_next_nbr]
    # fbd = fbd.at[self.bindex, ...].set(2 * f_nbr[self.bindex, ...] - f_next_nbr[self.bindex, ...])
    fbd = fbd.at[self.bindex, self.imissing].set(2 * f_nbr[self.bindex, self.imissing] - f_next_nbr[self.bindex, self.imissing])
    return fbd

jax_lab.core.boundary_conditions.NonEquilibriumExtrapolation

Bases: BoundaryCondition

Non-equilibrium extrapolation boundary condition.

Attributes

name (str): The name of the boundary condition. For this class, it is “NonEquilibriumExtrapolation”.

References

  1. Zhao-Li, G., Chu-Guang, Z. & Bao-Chang, S. Non-equilibrium extrapolation method for velocity and pressure boundary conditions in the lattice Boltzmann method. Chinese Phys. 11, 366-374 (2002).
Source code in jax_lab/core/boundary_conditions.py
class NonEquilibriumExtrapolation(BoundaryCondition):
    """
    Non-equilibrium extrapolation boundary condition.

    Attributes
    ----------
    name (str): The name of the boundary condition. For this class, it is "NonEquilibriumExtrapolation".

    References
    ----------
    1. Zhao-Li, G., Chu-Guang, Z. & Bao-Chang, S. Non-equilibrium extrapolation method for velocity and pressure boundary conditions in the lattice
    Boltzmann method. Chinese Phys. 11, 366-374 (2002).
    """

    def __init__(self, indices, grid_info, precision_policy, prescribed):
        super().__init__(indices, grid_info, precision_policy)
        self.name = "NonEquilibriumExtrapolation"
        self.needs_extra_configuration = False
        self.prescribed = prescribed
        # self.G_ff = self.compute_ff_greens_function()
        self.neighbors_found = False

    def find_neighbors(self):
        """
        Locate the nearest neighbouring fluid site (one per boundary node, along the inward normal) used to
        extrapolate the non-equilibrium part of the distribution. Must be called after self.normals is available,
        i.e. after BoundaryCondition.create_local_mask_and_normal_arrays has run.
        """
        ind = np.array(self.indices).T - self.normals
        self.indices_nbr = tuple(ind.T)

    @partial(jit, static_argnums=(0,))
    def apply(self, fout, _):
        """
        Applies the non-equilibrium extrapolation boundary condition. Density and velocity are evaluated only at
        the boundary and neighbor nodes (not over the whole domain) before being indexed.

        Parameters
        ----------
        fout (jax.numpy.ndarray): The output distribution functions.

        _ (jax.numpy.ndarray): The input distribution functions, not used in this function

        Returns
        -------
        jax.numpy.ndarray
            The modified output distribution functions after applying the boundary condition.
        """
        if not self.neighbors_found:
            self.find_neighbors()
            self.neighbors_found = True

        nbd = len(self.indices[0])
        bindex = np.arange(nbd)[:, None]
        fbd = fout[self.indices]
        f_nbr = fout[self.indices_nbr]

        rho_nbr = jnp.sum(f_nbr, axis=-1, keepdims=True)
        vel_nbr = jnp.dot(f_nbr, self.precision_policy.cast_to_compute(self.lattice.c.T)) / rho_nbr
        feq_nbr = self.equilibrium(rho_nbr, vel_nbr)
        feq = self.equilibrium(self.prescribed, vel_nbr)
        fneq_nbr = f_nbr - feq_nbr
        fbd = fbd.at[bindex, self.imissing].set(feq[bindex, self.imissing] + fneq_nbr[bindex, self.imissing])

        return fbd

apply

apply(fout, _)

Applies the non-equilibrium extrapolation boundary condition. Density and velocity are evaluated only at the boundary and neighbor nodes (not over the whole domain) before being indexed.

Parameters

fout (jax.numpy.ndarray): The output distribution functions.

_ (jax.numpy.ndarray): The input distribution functions, not used in this function

Returns

jax.numpy.ndarray The modified output distribution functions after applying the boundary condition.

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0,))
def apply(self, fout, _):
    """
    Applies the non-equilibrium extrapolation boundary condition. Density and velocity are evaluated only at
    the boundary and neighbor nodes (not over the whole domain) before being indexed.

    Parameters
    ----------
    fout (jax.numpy.ndarray): The output distribution functions.

    _ (jax.numpy.ndarray): The input distribution functions, not used in this function

    Returns
    -------
    jax.numpy.ndarray
        The modified output distribution functions after applying the boundary condition.
    """
    if not self.neighbors_found:
        self.find_neighbors()
        self.neighbors_found = True

    nbd = len(self.indices[0])
    bindex = np.arange(nbd)[:, None]
    fbd = fout[self.indices]
    f_nbr = fout[self.indices_nbr]

    rho_nbr = jnp.sum(f_nbr, axis=-1, keepdims=True)
    vel_nbr = jnp.dot(f_nbr, self.precision_policy.cast_to_compute(self.lattice.c.T)) / rho_nbr
    feq_nbr = self.equilibrium(rho_nbr, vel_nbr)
    feq = self.equilibrium(self.prescribed, vel_nbr)
    fneq_nbr = f_nbr - feq_nbr
    fbd = fbd.at[bindex, self.imissing].set(feq[bindex, self.imissing] + fneq_nbr[bindex, self.imissing])

    return fbd

find_neighbors

find_neighbors()

Locate the nearest neighbouring fluid site (one per boundary node, along the inward normal) used to extrapolate the non-equilibrium part of the distribution. Must be called after self.normals is available, i.e. after BoundaryCondition.create_local_mask_and_normal_arrays has run.

Source code in jax_lab/core/boundary_conditions.py
def find_neighbors(self):
    """
    Locate the nearest neighbouring fluid site (one per boundary node, along the inward normal) used to
    extrapolate the non-equilibrium part of the distribution. Must be called after self.normals is available,
    i.e. after BoundaryCondition.create_local_mask_and_normal_arrays has run.
    """
    ind = np.array(self.indices).T - self.normals
    self.indices_nbr = tuple(ind.T)

jax_lab.core.boundary_conditions.ExactNonEquilibriumExtrapolation

Bases: BoundaryCondition

Non-equilibrium extrapolation boundary condition but with added correction step to correct the density at the boundary node.

Attributes

name (str): The name of the boundary condition. For this class, it is “ExactNonEquilibriumExtrapolation”.

References

  1. Zhao-Li, G., Chu-Guang, Z. & Bao-Chang, S. Non-equilibrium extrapolation method for velocity and pressure boundary conditions in the lattice Boltzmann method. Chinese Phys. 11, 366-374 (2002).

  2. Fei, L., Qin, F., Zhao, J., Derome, D. & Carmeliet, J. Lattice Boltzmann modelling of isothermal two-component evaporation in porous media. Journal of Fluid Mechanics 955, A18 (2023).

Source code in jax_lab/core/boundary_conditions.py
class ExactNonEquilibriumExtrapolation(BoundaryCondition):
    """
    Non-equilibrium extrapolation boundary condition but with added correction step to correct the density at the boundary node.

    Attributes
    ----------
    name (str): The name of the boundary condition. For this class, it is "ExactNonEquilibriumExtrapolation".

    References
    ----------
    1. Zhao-Li, G., Chu-Guang, Z. & Bao-Chang, S. Non-equilibrium extrapolation method for velocity and pressure boundary conditions in the lattice
    Boltzmann method. Chinese Phys. 11, 366-374 (2002).

    2. Fei, L., Qin, F., Zhao, J., Derome, D. & Carmeliet, J. Lattice Boltzmann modelling of isothermal two-component evaporation in porous media.
    Journal of Fluid Mechanics 955, A18 (2023).
    """

    def __init__(self, indices, grid_info, precision_policy, prescribed, bc_type):
        super().__init__(indices, grid_info, precision_policy)
        self.name = "ExactNonEquilibriumExtrapolation"
        self.needs_extra_configuration = False
        self.prescribed = prescribed
        self.type = bc_type
        self.w_NEQ = self.compute_NEQ_weights()
        self.neighbors_found = False

    def find_neighbors(self):
        """
        Locate the nearest neighbouring fluid site (one per boundary node, along the inward normal) used to
        extrapolate the non-equilibrium part of the distribution. Must be called after self.normals is available,
        i.e. after BoundaryCondition.create_local_mask_and_normal_arrays has run.
        """
        ind = np.array(self.indices).T - self.normals
        self.indices_nbr = tuple(ind.T)

    def compute_NEQ_weights(self):
        """
        Define the weight function used to correct the distribution values obtained Non-Equilibrium extrapolation boundary condition.

        Typically, it is defined as:

        w(i) = g1 if |x - x'| = 1
             = g2 if |x - x'| = sqrt(2)
             = 0, otherwise

        Here d is the dimension of problem and x' are the neighboring points.

        Some examples values could be:
        For D2Q9:
            g1 = 1/3 and g2 = 1/12
        For D3Q19
            g1 = 1/6 and g2 = 1/12

        Returns
        -------
        G_ff: jax.numpy.ndarray.
            Dimension: (q, )
        """
        c = np.array(self.lattice.c).T
        w_NEQ = np.zeros((self.lattice.q,), dtype=np.float64)
        cl = np.linalg.norm(c, axis=-1)
        if isinstance(self.lattice, LatticeD2Q9):
            g1 = 1.0 / 3.0
            g2 = 1.0 / 12.0
            w_NEQ[np.isclose(cl, 1.0, atol=1e-6)] = g1
            w_NEQ[np.isclose(cl, jnp.sqrt(2.0), atol=1e-6)] = g2
        elif isinstance(self.lattice, LatticeD3Q19):
            g1 = 1.0 / 6.0
            g2 = 1.0 / 12.0
            w_NEQ[np.isclose(cl, 1.0, atol=1e-6)] = g1
            w_NEQ[np.isclose(cl, jnp.sqrt(2.0), atol=1e-6)] = g2
        return jnp.array(w_NEQ, dtype=self.precision_policy.compute_dtype)

    @partial(jit, static_argnums=(0,))
    def apply(self, fout, _):
        """
        Applies the non-equilibrium extrapolation boundary condition. Density and velocity are evaluated only at
        the boundary and neighbor nodes (not over the whole domain) before being indexed.

        Parameters
        ----------
        fout (jax.numpy.ndarray): The output distribution functions.

        _ (jax.numpy.ndarray): The input distribution functions, not used in this function

        Returns
        -------
        (jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.
        """
        if not self.neighbors_found:
            self.find_neighbors()
            self.neighbors_found = True

        nbd = len(self.indices[0])
        bindex = np.arange(nbd)[:, None]
        fbd = fout[self.indices]
        f_nbr = fout[self.indices_nbr]

        rho_nbr = jnp.sum(f_nbr, axis=-1, keepdims=True)
        vel_nbr = jnp.dot(f_nbr, self.precision_policy.cast_to_compute(self.lattice.c.T)) / rho_nbr
        feq_nbr = self.equilibrium(rho_nbr, vel_nbr)
        feq = self.equilibrium(self.prescribed, vel_nbr)
        fneq_nbr = f_nbr - feq_nbr
        fbd = fbd.at[bindex, self.imissing].set(feq[bindex, self.imissing] + fneq_nbr[bindex, self.imissing])

        # Correction step: redistribute the density error over the unknown (imissing) directions only, weighted
        # by w_NEQ, so that the corrected boundary density matches self.prescribed exactly.
        rho_incorrect = jnp.sum(fbd, axis=-1, keepdims=True)
        w_missing = self.w_NEQ * self.imissing_mask
        beta = w_missing * (self.prescribed - rho_incorrect) / jnp.sum(w_missing, axis=-1, keepdims=True)
        fbd = fbd.at[bindex, self.imissing].set(fbd[bindex, self.imissing] + beta[bindex, self.imissing])
        return fbd

apply

apply(fout, _)

Applies the non-equilibrium extrapolation boundary condition. Density and velocity are evaluated only at the boundary and neighbor nodes (not over the whole domain) before being indexed.

Parameters

fout (jax.numpy.ndarray): The output distribution functions.

_ (jax.numpy.ndarray): The input distribution functions, not used in this function

Returns

(jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0,))
def apply(self, fout, _):
    """
    Applies the non-equilibrium extrapolation boundary condition. Density and velocity are evaluated only at
    the boundary and neighbor nodes (not over the whole domain) before being indexed.

    Parameters
    ----------
    fout (jax.numpy.ndarray): The output distribution functions.

    _ (jax.numpy.ndarray): The input distribution functions, not used in this function

    Returns
    -------
    (jax.numpy.ndarray): The modified output distribution functions after applying the boundary condition.
    """
    if not self.neighbors_found:
        self.find_neighbors()
        self.neighbors_found = True

    nbd = len(self.indices[0])
    bindex = np.arange(nbd)[:, None]
    fbd = fout[self.indices]
    f_nbr = fout[self.indices_nbr]

    rho_nbr = jnp.sum(f_nbr, axis=-1, keepdims=True)
    vel_nbr = jnp.dot(f_nbr, self.precision_policy.cast_to_compute(self.lattice.c.T)) / rho_nbr
    feq_nbr = self.equilibrium(rho_nbr, vel_nbr)
    feq = self.equilibrium(self.prescribed, vel_nbr)
    fneq_nbr = f_nbr - feq_nbr
    fbd = fbd.at[bindex, self.imissing].set(feq[bindex, self.imissing] + fneq_nbr[bindex, self.imissing])

    # Correction step: redistribute the density error over the unknown (imissing) directions only, weighted
    # by w_NEQ, so that the corrected boundary density matches self.prescribed exactly.
    rho_incorrect = jnp.sum(fbd, axis=-1, keepdims=True)
    w_missing = self.w_NEQ * self.imissing_mask
    beta = w_missing * (self.prescribed - rho_incorrect) / jnp.sum(w_missing, axis=-1, keepdims=True)
    fbd = fbd.at[bindex, self.imissing].set(fbd[bindex, self.imissing] + beta[bindex, self.imissing])
    return fbd

compute_NEQ_weights

compute_NEQ_weights()

Define the weight function used to correct the distribution values obtained Non-Equilibrium extrapolation boundary condition.

Typically, it is defined as:

w(i) = g1 if |x - x’| = 1 = g2 if |x - x’| = sqrt(2) = 0, otherwise

Here d is the dimension of problem and x’ are the neighboring points.

Some examples values could be: For D2Q9: g1 = 1/3 and g2 = 1/12 For D3Q19 g1 = 1/6 and g2 = 1/12

Returns

G_ff: jax.numpy.ndarray. Dimension: (q, )

Source code in jax_lab/core/boundary_conditions.py
def compute_NEQ_weights(self):
    """
    Define the weight function used to correct the distribution values obtained Non-Equilibrium extrapolation boundary condition.

    Typically, it is defined as:

    w(i) = g1 if |x - x'| = 1
         = g2 if |x - x'| = sqrt(2)
         = 0, otherwise

    Here d is the dimension of problem and x' are the neighboring points.

    Some examples values could be:
    For D2Q9:
        g1 = 1/3 and g2 = 1/12
    For D3Q19
        g1 = 1/6 and g2 = 1/12

    Returns
    -------
    G_ff: jax.numpy.ndarray.
        Dimension: (q, )
    """
    c = np.array(self.lattice.c).T
    w_NEQ = np.zeros((self.lattice.q,), dtype=np.float64)
    cl = np.linalg.norm(c, axis=-1)
    if isinstance(self.lattice, LatticeD2Q9):
        g1 = 1.0 / 3.0
        g2 = 1.0 / 12.0
        w_NEQ[np.isclose(cl, 1.0, atol=1e-6)] = g1
        w_NEQ[np.isclose(cl, jnp.sqrt(2.0), atol=1e-6)] = g2
    elif isinstance(self.lattice, LatticeD3Q19):
        g1 = 1.0 / 6.0
        g2 = 1.0 / 12.0
        w_NEQ[np.isclose(cl, 1.0, atol=1e-6)] = g1
        w_NEQ[np.isclose(cl, jnp.sqrt(2.0), atol=1e-6)] = g2
    return jnp.array(w_NEQ, dtype=self.precision_policy.compute_dtype)

find_neighbors

find_neighbors()

Locate the nearest neighbouring fluid site (one per boundary node, along the inward normal) used to extrapolate the non-equilibrium part of the distribution. Must be called after self.normals is available, i.e. after BoundaryCondition.create_local_mask_and_normal_arrays has run.

Source code in jax_lab/core/boundary_conditions.py
def find_neighbors(self):
    """
    Locate the nearest neighbouring fluid site (one per boundary node, along the inward normal) used to
    extrapolate the non-equilibrium part of the distribution. Must be called after self.normals is available,
    i.e. after BoundaryCondition.create_local_mask_and_normal_arrays has run.
    """
    ind = np.array(self.indices).T - self.normals
    self.indices_nbr = tuple(ind.T)

jax_lab.core.boundary_conditions.ConvectiveOutflow

Bases: BoundaryCondition

Extrapolation outflow boundary condition for a lattice Boltzmann method simulation.

This class implements the extrapolation outflow boundary condition, which is a type of outflow boundary condition that uses extrapolation to avoid strong wave reflections.

Attributes

name (str): The name of the boundary condition. For this class, it is “ConvectiveOutflow”.

References

  1. Lou, Q., Guo, Z. & Shi, B. Evaluation of outflow boundary conditions for two-phase lattice Boltzmann equation. Phys. Rev. E 87, 063301 (2013). doi: doi.org/10.1103/PhysRevE.87.063301
Source code in jax_lab/core/boundary_conditions.py
class ConvectiveOutflow(BoundaryCondition):
    """
    Extrapolation outflow boundary condition for a lattice Boltzmann method simulation.

    This class implements the extrapolation outflow boundary condition, which is a type of outflow boundary condition
    that uses extrapolation to avoid strong wave reflections.

    Attributes
    ----------
    name (str): The name of the boundary condition. For this class, it is "ConvectiveOutflow".

    References
    ----------
    1. Lou, Q., Guo, Z. & Shi, B. Evaluation of outflow boundary conditions for two-phase lattice Boltzmann equation.
    Phys. Rev. E 87, 063301 (2013). doi: doi.org/10.1103/PhysRevE.87.063301
    """

    def __init__(self, indices, grid_info, precision_policy):
        super().__init__(indices, grid_info, precision_policy)
        self.name = "ConvectiveOutflow"
        self.needs_extra_configuration = False
        self.neighbors_found = False

    def configure(self, boundary_mask):
        """
        Correct boundary indices to ensure that only voxelized surfaces with normal vectors along main cartesian axes
        are assigned this type of BC.
        """
        nv = np.dot(self.lattice.c, ~boundary_mask.T)
        corner_voxels = np.count_nonzero(nv, axis=0) > 1
        self.indices = tuple(np.array(self.indices)[:, ~corner_voxels])

    def find_neighbors(self):
        ind = np.array(self.indices).T - self.normals
        self.indices_nbr = tuple(ind.T)
        nbd = len(self.indices[0])
        self.bindex = np.arange(nbd)[:, None]

    @partial(jit, static_argnums=(0,))
    def apply(self, fout, fin):
        """
        Applies the convective outflow boundary condition.

        Parameters
        ----------
        fout (jax.numpy.ndarray): The output distribution functions.

        fin (jax.numpy.ndarray): The input distribution functions.

        Returns
        -------
        jax.numpy.ndarray
            The modified output distribution functions after applying the boundary condition.
        """
        if not self.neighbors_found:
            self.find_neighbors()
            self.neighbors_found = True

        nbd = len(self.indices[0])
        bindex = np.arange(nbd)[:, None]

        f_nbr = fout[self.indices_nbr]
        rho_nbr = jnp.sum(f_nbr, axis=-1, keepdims=True)
        u_nbr = jnp.sum((jnp.dot(f_nbr, self.lattice.c.T) / rho_nbr) * self.normals, axis=-1, keepdims=True)
        lambda_cbc = jnp.max(u_nbr)

        # Retrieve previous timestep's imissing values (stored in iknown slots during PostCollision)
        f_prev_missing = fin[self.indices][bindex, self.iknown]

        # Start from post-streaming values (correct for known directions)
        # fbd = fout[self.indices]
        # Apply convective formula to missing directions only
        # f_new_missing = (f_prev_missing + lambda_cbc * f_nbr[bindex, self.imissing]) / (1 + lambda_cbc)
        # f_new_missing = (1 - lambda_cbc) * f_prev_missing + lambda_cbc * f_nbr[bindex, self.imissing]
        # fbd = fbd.at[bindex, self.imissing].set(f_new_missing)

        f_prev_missing = fin[self.indices]
        # fbd = (f_prev_missing + lambda_cbc * f_nbr) / (1 + lambda_cbc)
        fbd = (1 - lambda_cbc) * f_nbr + lambda_cbc * f_prev_missing

        return fbd

apply

apply(fout, fin)

Applies the convective outflow boundary condition.

Parameters

fout (jax.numpy.ndarray): The output distribution functions.

fin (jax.numpy.ndarray): The input distribution functions.

Returns

jax.numpy.ndarray The modified output distribution functions after applying the boundary condition.

Source code in jax_lab/core/boundary_conditions.py
@partial(jit, static_argnums=(0,))
def apply(self, fout, fin):
    """
    Applies the convective outflow boundary condition.

    Parameters
    ----------
    fout (jax.numpy.ndarray): The output distribution functions.

    fin (jax.numpy.ndarray): The input distribution functions.

    Returns
    -------
    jax.numpy.ndarray
        The modified output distribution functions after applying the boundary condition.
    """
    if not self.neighbors_found:
        self.find_neighbors()
        self.neighbors_found = True

    nbd = len(self.indices[0])
    bindex = np.arange(nbd)[:, None]

    f_nbr = fout[self.indices_nbr]
    rho_nbr = jnp.sum(f_nbr, axis=-1, keepdims=True)
    u_nbr = jnp.sum((jnp.dot(f_nbr, self.lattice.c.T) / rho_nbr) * self.normals, axis=-1, keepdims=True)
    lambda_cbc = jnp.max(u_nbr)

    # Retrieve previous timestep's imissing values (stored in iknown slots during PostCollision)
    f_prev_missing = fin[self.indices][bindex, self.iknown]

    # Start from post-streaming values (correct for known directions)
    # fbd = fout[self.indices]
    # Apply convective formula to missing directions only
    # f_new_missing = (f_prev_missing + lambda_cbc * f_nbr[bindex, self.imissing]) / (1 + lambda_cbc)
    # f_new_missing = (1 - lambda_cbc) * f_prev_missing + lambda_cbc * f_nbr[bindex, self.imissing]
    # fbd = fbd.at[bindex, self.imissing].set(f_new_missing)

    f_prev_missing = fin[self.indices]
    # fbd = (f_prev_missing + lambda_cbc * f_nbr) / (1 + lambda_cbc)
    fbd = (1 - lambda_cbc) * f_nbr + lambda_cbc * f_prev_missing

    return fbd

configure

configure(boundary_mask)

Correct boundary indices to ensure that only voxelized surfaces with normal vectors along main cartesian axes are assigned this type of BC.

Source code in jax_lab/core/boundary_conditions.py
def configure(self, boundary_mask):
    """
    Correct boundary indices to ensure that only voxelized surfaces with normal vectors along main cartesian axes
    are assigned this type of BC.
    """
    nv = np.dot(self.lattice.c, ~boundary_mask.T)
    corner_voxels = np.count_nonzero(nv, axis=0) > 1
    self.indices = tuple(np.array(self.indices)[:, ~corner_voxels])

jax_lab.core.boundary_conditions.ThermalBoundaryCondition

Bases: object

Base class for thermal (temperature field) boundary conditions used by the hybrid thermal solver in thermal.py.

Unlike the LBM boundary conditions above which act on distribution functions, thermal boundary conditions act directly on the temperature field of the finite difference solver.

Parameters

indices (tuple of numpy.ndarray): Tuple of index arrays selecting the boundary nodes, one array per spatial axis (e.g. tuple(wall_indices.T)).

Source code in jax_lab/core/boundary_conditions.py
class ThermalBoundaryCondition(object):
    """
    Base class for thermal (temperature field) boundary conditions used by the
    hybrid thermal solver in thermal.py.

    Unlike the LBM boundary conditions above which act on distribution
    functions, thermal boundary conditions act directly on the temperature
    field of the finite difference solver.

    Parameters
    ----------
    indices (tuple of numpy.ndarray): Tuple of index arrays selecting the boundary
    nodes, one array per spatial axis (e.g. tuple(wall_indices.T)).
    """

    def __init__(self, indices):
        self.indices = tuple(np.asarray(idx) for idx in indices)
        self.name = None

    def apply(self, T, timestep):
        """
        Apply the boundary condition to the temperature field.

        Parameters
        ----------
        T (jax.numpy.ndarray): Temperature field of shape (nx, ny, 1) in 2D or
        (nx, ny, nz, 1) in 3D.

        timestep (int): Current timestep, available for time dependent conditions.

        Returns
        -------
        jax.numpy.ndarray: Temperature field with the boundary condition applied.
        """
        raise NotImplementedError

apply

apply(T, timestep)

Apply the boundary condition to the temperature field.

Parameters

T (jax.numpy.ndarray): Temperature field of shape (nx, ny, 1) in 2D or (nx, ny, nz, 1) in 3D.

timestep (int): Current timestep, available for time dependent conditions.

Returns

jax.numpy.ndarray: Temperature field with the boundary condition applied.

Source code in jax_lab/core/boundary_conditions.py
def apply(self, T, timestep):
    """
    Apply the boundary condition to the temperature field.

    Parameters
    ----------
    T (jax.numpy.ndarray): Temperature field of shape (nx, ny, 1) in 2D or
    (nx, ny, nz, 1) in 3D.

    timestep (int): Current timestep, available for time dependent conditions.

    Returns
    -------
    jax.numpy.ndarray: Temperature field with the boundary condition applied.
    """
    raise NotImplementedError

jax_lab.core.boundary_conditions.DirichletTemperature

Bases: ThermalBoundaryCondition

Dirichlet (prescribed temperature) boundary condition: T = T_w at the boundary nodes.

Parameters

indices (tuple of numpy.ndarray): Index arrays of the boundary nodes.

prescribed (float or numpy.ndarray): Prescribed wall temperature. Either a scalar applied to all nodes or an array of shape (n, 1) with one value per boundary node.

Source code in jax_lab/core/boundary_conditions.py
class DirichletTemperature(ThermalBoundaryCondition):
    """
    Dirichlet (prescribed temperature) boundary condition: T = T_w at the
    boundary nodes.

    Parameters
    ----------
    indices (tuple of numpy.ndarray): Index arrays of the boundary nodes.

    prescribed (float or numpy.ndarray): Prescribed wall temperature. Either a
    scalar applied to all nodes or an array of shape (n, 1) with one value per
    boundary node.
    """

    def __init__(self, indices, prescribed):
        super().__init__(indices)
        self.name = "DirichletTemperature"
        self.prescribed = prescribed

    def apply(self, T, timestep):
        return T.at[self.indices].set(self.prescribed)

jax_lab.core.boundary_conditions.NeumannTemperature

Bases: ThermalBoundaryCondition

Neumann (prescribed normal temperature gradient) boundary condition, imposed with a first order one-sided difference over unit spacing: T_wall = T_interior + q, where q = dT/dn is the prescribed gradient along the outward normal (q = 0 gives an adiabatic wall).

Assumes the interior neighbor of every boundary node lies one node along the negated outward normal. Corner nodes shared with a Dirichlet boundary should be listed in the Dirichlet condition as well, appended after this one, so the Dirichlet value takes precedence.

Parameters

indices (tuple of numpy.ndarray): Index arrays of the boundary nodes.

normal (sequence of int): Outward unit normal of the boundary, e.g. (0, 1) for the top wall in 2D or (0, 0, -1) for the bottom wall in 3D.

prescribed (float or numpy.ndarray): Prescribed outward normal gradient. Either a scalar or an array of shape (n, 1). Defaults to 0 (adiabatic).

Source code in jax_lab/core/boundary_conditions.py
class NeumannTemperature(ThermalBoundaryCondition):
    """
    Neumann (prescribed normal temperature gradient) boundary condition,
    imposed with a first order one-sided difference over unit spacing:
    T_wall = T_interior + q, where q = dT/dn is the prescribed gradient along
    the outward normal (q = 0 gives an adiabatic wall).

    Assumes the interior neighbor of every boundary node lies one node along
    the negated outward normal. Corner nodes shared with a Dirichlet boundary
    should be listed in the Dirichlet condition as well, appended after this
    one, so the Dirichlet value takes precedence.

    Parameters
    ----------
    indices (tuple of numpy.ndarray): Index arrays of the boundary nodes.

    normal (sequence of int): Outward unit normal of the boundary, e.g. (0, 1)
    for the top wall in 2D or (0, 0, -1) for the bottom wall in 3D.

    prescribed (float or numpy.ndarray): Prescribed outward normal gradient.
    Either a scalar or an array of shape (n, 1). Defaults to 0 (adiabatic).
    """

    def __init__(self, indices, normal, prescribed=0.0):
        super().__init__(indices)
        self.name = "NeumannTemperature"
        self.prescribed = prescribed
        self.neighbor_indices = tuple(np.asarray(idx) - int(n) for idx, n in zip(self.indices, normal))

    def apply(self, T, timestep):
        return T.at[self.indices].set(T[self.neighbor_indices] + self.prescribed)