Tools

from dedalus.tools.config import config

# ~/.dedalus/dedalus.cfg

[parallelism]  
TRANSPOSE_LIBRARY = fftw
GROUP_TRANSPOSES = True

[linear algebra]
MATRIX_FACTORIZER = SuperluNaturalSpsolve

[memory]  
FIELD_CACHE_SIZE = 128

[logging]
stdout_level = info
file_level = debug

from dedalus.tools.config import config

# Get values
transpose_lib = config['parallelism'].get('TRANSPOSE_LIBRARY')
cache_size = config['memory'].getint('FIELD_CACHE_SIZE')  

# Check boolean
group_transposes = config['parallelism'].getboolean('GROUP_TRANSPOSES')

from dedalus.tools.parallel import Sync

from dedalus.tools.parallel import Sync  
from mpi4py import MPI
import time

comm = MPI.COMM_WORLD

# Synchronize before and after block
with Sync(comm):
    # All processes wait here  
    time.sleep(comm.rank * 0.1)
    print(f"Process {comm.rank}")
# All processes wait here before continuing

from dedalus.tools.parallel import parallel_mkdir

from dedalus.tools.parallel import parallel_mkdir
import pathlib

output_dir = pathlib.Path('output')  
parallel_mkdir(output_dir)

from dedalus.tools import post  

from dedalus.tools import post

# Merge output from parallel run
post.merge_process_files('snapshots', cleanup=True)  
post.merge_process_files('analysis', cleanup=False)

import h5py  
import numpy as np
import matplotlib.pyplot as plt
from dedalus.tools import post

# Merge files
post.merge_process_files('snapshots', cleanup=True)

# Read merged data  
with h5py.File('snapshots/snapshots_s1.h5', 'r') as f:
    # Load data
    u = f['tasks']['velocity'][:]
    x = f['tasks']['velocity'].dims[1][0][:]  
    t = f['scales']['sim_time'][:]
    
    # Plot
    plt.figure(figsize=(10, 6))
    for i in range(0, len(t), 10):
        plt.plot(x, u[i], label=f't={t[i]:.2f}')  
    plt.xlabel('x')
    plt.ylabel('u')
    plt.legend()
    plt.savefig('evolution.png', dpi=150)

from dedalus.tools.array import reshape_vector

import numpy as np
from dedalus.tools.array import reshape_vector

x = np.linspace(0, 1, 64)

# Reshape for broadcasting  
x_2d = reshape_vector(x, dim=2, axis=0)  # Shape: (64, 1)
x_3d = reshape_vector(x, dim=3, axis=1)  # Shape: (1, 64, 1)

from dedalus.tools.array import axslice

import numpy as np  
from dedalus.tools.array import axslice

data = np.random.rand(10, 20, 30)

# Slice along axis 1
sliced = data[axslice(1, 5, 15)]  # data[:, 5:15, :]

from dedalus.tools.exceptions import *

import dedalus.public as d3  
from dedalus.tools.exceptions import UnsupportedEquationError

try:
    problem = d3.LBVP([u])
    # This will raise UnsupportedEquationError (nonlinear RHS)  
    problem.add_equation("dx(u) = u**2")
except UnsupportedEquationError as e:
    print(f"Invalid equation: {e}")

import logging
logger = logging.getLogger(__name__)

# In dedalus.cfg
[logging]
stdout_level = info
file_level = debug
filename = dedalus.log

import logging
logger = logging.getLogger(__name__)

# Set level for your script
logger.setLevel(logging.INFO)

# Log messages  
logger.debug("Detailed debug information")
logger.info("Iteration %d complete" % iteration)
logger.warning("CFL condition violated")
logger.error("Solver failed to converge")  

from dedalus.tools.cache import CachedMethod  

from dedalus.tools.cache import CachedMethod

class MyClass:  
    @CachedMethod
    def expensive_computation(self, x):
        # Only computed once per unique x
        return x**2 + 2*x + 1

obj = MyClass()  
result1 = obj.expensive_computation(5)  # Computed
result2 = obj.expensive_computation(5)  # Cached

from dedalus.tools.cache import CachedAttribute

from dedalus.tools.cache import CachedAttribute  
import numpy as np

class MyClass:
    @CachedAttribute
    def random_array(self):
        # Only generated once
        return np.random.rand(1000)  

obj = MyClass()
arr1 = obj.random_array  # Generated
arr2 = obj.random_array  # Same array

from dedalus.tools.general import unify

from dedalus.tools.general import unify

# Returns common value if all equal
result = unify([1, 1, 1, 1])  # Returns 1  

# Raises error if not all equal
try:
    result = unify([1, 2, 3])
except ValueError:
    print("Values not unified")  

from dedalus.tools.general import OrderedSet  

from dedalus.tools.general import OrderedSet

s = OrderedSet([3, 1, 4, 1, 5])  
print(list(s))  # [3, 1, 4, 5]

from dedalus.tools.random_arrays import ChunkedRandomArray

from dedalus.tools.random_arrays import ChunkedRandomArray  
import numpy as np

# Create reproducible random array across processes
shape = (64, 64, 64)
seed = 42
random_gen = ChunkedRandomArray(shape, seed=seed)  
local_data = random_gen.get_local_data()

# In dedalus.cfg  
[profiling]
PROFILE_DEFAULT = False
PARALLEL_PROFILE_DEFAULT = False
PROFILE_DIRECTORY = profiles

import dedalus.public as d3

# Enable profiling for solver
solver = problem.build_solver(d3.RK443)
solver.profile = True

# Run simulation  
while solver.proceed:
    solver.step(dt)

# Profile data saved to profiles/

from dedalus.extras import quick_domains

from dedalus.extras import quick_domains

# Quick 2D channel setup
coords, dist, (xbasis, ybasis) = quick_domains.channel_2d(N=64, dealias=3/2)

# Create fields
u = dist.VectorField(coords, bases=(xbasis, ybasis))
p = dist.Field(bases=(xbasis, ybasis))

# Ready to set up problem
problem = d3.IVP([u, p], namespace=locals())