Skip to content
Snippets Groups Projects
  • Martin Bauer's avatar
    41f3e83a
    Separated modules into subfolders with own setup.py · 41f3e83a
    Martin Bauer authored
    This restructuring allows for easier separation of modules into
    separate repositories later. Also, now pip install with repo url can be
    used.
    
    The setup.py files have also been updated to correctly reference each
    other. Module versions are not extracted from git state
    41f3e83a
    History
    Separated modules into subfolders with own setup.py
    Martin Bauer authored
    This restructuring allows for easier separation of modules into
    separate repositories later. Also, now pip install with repo url can be
    used.
    
    The setup.py files have also been updated to correctly reference each
    other. Module versions are not extracted from git state
test_lbstep.py 4.04 KiB
import pytest
import numpy as np
from lbmpy.scenarios import create_lid_driven_cavity, create_fully_periodic_flow

try:
    import pycuda.driver
    gpu_available = True
except ImportError:
    gpu_available = False

try:
    import waLBerla as wLB
    parallel_available = wLB.cpp_available
except ImportError:
    parallel_available = False
    wLB = None


def ldc_setup(**kwargs):
    ldc = create_lid_driven_cavity(relaxation_rate=1.7, **kwargs)
    ldc.run(50)
    return ldc.density_slice()


def test_data_handling_3d():
    print("--- LDC 3D test ---")
    results = []
    for parallel in [False, True] if parallel_available else [False]:
        for gpu in [False, True] if gpu_available else [False]:
            if parallel and gpu and not hasattr(wLB, 'cuda'):
                continue
            print("Testing parallel: %s\tgpu: %s" % (parallel, gpu))
            opt_params = {'target': 'gpu' if gpu else 'cpu',
                          'gpu_indexing_params': {'block_size': (8, 4, 2)}}
            if parallel:
                from pystencils.datahandling import ParallelDataHandling
                blocks = wLB.createUniformBlockGrid(blocks=(2, 3, 4), cellsPerBlock=(5, 5, 5),
                                                    oneBlockPerProcess=False)
                dh = ParallelDataHandling(blocks, dim=3)
                rho = ldc_setup(data_handling=dh, optimization=opt_params)
                results.append(rho)
            else:
                rho = ldc_setup(domain_size=(10, 15, 20), parallel=False, optimization=opt_params)
                results.append(rho)
    for i, arr in enumerate(results[1:]):
        print("Testing equivalence version 0 with version %d" % (i + 1,))
        np.testing.assert_almost_equal(results[0], arr)


def test_data_handling_2d():
    print("--- LDC 2D test ---")
    results = []
    for parallel in [True, False] if parallel_available else [False]:
        for gpu in [True, False] if gpu_available else [False]:
            if parallel and gpu and not hasattr(wLB, 'cuda'):
                continue

            print("Testing parallel: %s\tgpu: %s" % (parallel, gpu))
            opt_params = {'target': 'gpu' if gpu else 'cpu',
                          'gpu_indexing_params': {'block_size': (8, 4, 2)}}
            if parallel:
                from pystencils.datahandling import ParallelDataHandling
                blocks = wLB.createUniformBlockGrid(blocks=(2, 3, 1), cellsPerBlock=(5, 5, 1),
                                                    oneBlockPerProcess=False)
                dh = ParallelDataHandling(blocks, dim=2)
                rho = ldc_setup(data_handling=dh, optimization=opt_params)
                results.append(rho)
            else:
                rho = ldc_setup(domain_size=(10, 15), parallel=False, optimization=opt_params)
                results.append(rho)
    for i, arr in enumerate(results[1:]):
        print("Testing equivalence version 0 with version %d" % (i + 1,))
        np.testing.assert_almost_equal(results[0], arr)


def test_smagorinsky_setup():
    step = create_lid_driven_cavity((30, 30), smagorinsky=0.16, relaxation_rate=1.99)
    step.run(10)


def test_advanced_initialization():
    width, height = 100, 50
    velocity_magnitude = 0.05
    init_vel = np.zeros((width, height, 2))
    # fluid moving to the right everywhere...
    init_vel[:, :, 0] = velocity_magnitude
    # ...except at a stripe in the middle, where it moves left
    init_vel[:, height // 3: height // 3 * 2, 0] = -velocity_magnitude
    # small random y velocity component
    init_vel[:, :, 1] = 0.1 * velocity_magnitude * np.random.rand(width, height)
    shear_flow_scenario = create_fully_periodic_flow(initial_velocity=init_vel, relaxation_rate=1.95)
    with pytest.raises(ValueError) as e:
        shear_flow_scenario.run_iterative_initialization(max_steps=20000, check_residuum_after=500)
    assert 'did not converge' in str(e)

    shear_flow_scenario = create_fully_periodic_flow(initial_velocity=init_vel, relaxation_rate=1.6)
    shear_flow_scenario.run_iterative_initialization(max_steps=20000, check_residuum_after=500)