pmmoto.core package

Initialize the core subpackage and import core modules for PMMoTo.

This package provides core domain, subdomain, and utility functionality.

Submodules

pmmoto.core.boundary_types module

boundary_types.py

Possible boundary types for PMMoTo

pmmoto.core.boundary_types.boundary_order(boundaries: list[BoundaryType]) → BoundaryType

Prove the boundary type for joining boundaries.

Fro edges and corners in 3D, this specifies the boundary type for the feature. #TODO Get rid of this and allow edges and corners to carry all types.

Parameters:: boundaries (list[BoundaryType]) – boundaries at given feature
Returns:: The one boundary type for the feature
Return type:: BoundaryType

pmmoto.core.communication module

communication.py

pmmoto.core.communication.all_gather(data: Any) → list[Any]

Gather data from all processes and return the combined result.

Parameters:: data – Data to be gathered from each process.
Returns:: List containing data from all processes.

pmmoto.core.communication.all_reduce(data: Any, op: str | Op = 'sum') → Any

Reduce data from all processes using given operation and distribute result.

Parameters:

data – Local data to reduce (e.g., int, float, NumPy array).
op – MPI operation or string alias (e.g., ‘sum’, ‘max’).

Returns:

The reduced result, available to all processes.

Raises:

ValueError – If the provided operation is not supported.

pmmoto.core.communication.communicate_features(subdomain: Subdomain | PaddedSubdomain | VerletSubdomain, send_data: dict[tuple[int, ...], Any]) → dict[tuple[int, ...], Any]

Send data between processes for faces, edges, and corners.

This also swaps the feature ids.

Parameters:

subdomain (object) – Subdomain object containing rank and features information.
send_data (dict) – The data to be sent to neighboring processes.

Returns:

Received data from neighboring processes.: If unpack is True, returns unpacked received data.

Return type:

dict

pmmoto.core.communication.gather(data: T, root: int = 0) → list[T] | None

Gather data from all processes to the root process.

Parameters:

data – Data to be gathered from each process.
root – Rank of the root process that receives the data.

Returns:

List of data on root process, None on others.

pmmoto.core.communication.update_buffer(subdomain: Subdomain | PaddedSubdomain | VerletSubdomain, img: ndarray[Any, dtype[T]]) → ndarray[Any, dtype[T]]

Update the buffer on subdomains based on their current feature info.

Parameters:

subdomain – The subdomain object.
img – The grid data to be updated.

Returns:

The updated grid.

pmmoto.core.domain module

domain.py

Defines the Domain class for representing the physical simulation domain in PMMoTo.

class pmmoto.core.domain.Domain(box: tuple[tuple[float, float], ...], boundary_types: tuple[tuple[BoundaryType, BoundaryType], ...] = ((BoundaryType.END, BoundaryType.END), (BoundaryType.END, BoundaryType.END), (BoundaryType.END, BoundaryType.END)), inlet: tuple[tuple[bool, bool], ...] = ((False, False), (False, False), (False, False)), outlet: tuple[tuple[int, int], ...] = ((False, False), (False, False), (False, False)))

Bases: object

Represent a physical simulation domain.

box

Physical bounds for each dimension.

Type:: tuple[tuple[float, float], …]

boundary_types

Boundary types for each face. END: No assumption made WALL: Wall boundary condition PERIODIC: Periodic boundary condition (opposing face must also be 2)

Type:: tuple[tuple[BoundaryType, BoundaryType], …]

inlet

Inlet flags (must be 0 boundary type).

Type:: tuple[tuple[bool, bool], …]

outlet

Outlet flags (must be 0 boundary type).

Type:: tuple[tuple[bool, bool], …]

dims

Number of spatial dimensions (default 3).

Type:: int

volume

Volume of the domain.

Type:: float

periodic

True if any boundary is periodic.

Type:: bool

length

Length of the domain in each dimension.

Type:: tuple[float, …]

Initialize a Domain.

Parameters:

box (tuple[tuple[float, float], ...]) – Physical bounds for each dimension.
boundary_types (tuple[tuple[int, int], ...], optional) – Boundary types.
inlet (tuple[tuple[int, int], ...], optional) – Inlet flags for each face.
outlet (tuple[tuple[int, int], ...], optional) – Outlet flags for each face.

get_length() → tuple[float, ...]

Calculate the length of the domain in each dimension.

Returns:: Length in each dimension.
Return type:: tuple[float, …]

get_origin() → tuple[float, ...]

Determine the domain origin from box.

Returns:: Domain origin.
Return type:: tuple[float, …]

get_volume() → floating[Any]

Calculate the volume of the domain.

Returns:: Volume of the domain.
Return type:: float

periodic_check() → bool

Check if any external boundary is a periodic boundary.

Returns:: True if any boundary is periodic, False otherwise.
Return type:: bool

pmmoto.core.domain_decompose module

decomposed_domain.py

Defines the DecomposedDomain class for dividing a discretized domain into subdomains.

class pmmoto.core.domain_decompose.DecomposedDomain(subdomains: tuple[int, ...] = (1, 1, 1), **kwargs: Any)

Bases: DiscretizedDomain

Collection of subdomains for domain decomposition.

Used to divide the domain into subdomains and pass properties to each subdomain.

Initialize a DecomposedDomain.

Parameters:

subdomains (tuple[int, int, int] optional) – Number of subdomains
**kwargs – Additional arguments passed to DiscretizedDomain.

classmethod from_discretized_domain(discretized_domain: DiscretizedDomain, subdomains: tuple[int, ...]) → Self

Create a DecomposedDomain from an existing DiscretizedDomain and subdomains.

Parameters:

discretized_domain (DiscretizedDomain) – The discretized domain object.
subdomains – Number of subdomains in each spatial dimension.

Returns:

New decomposed domain instance.

Return type:

DecomposedDomain

gen_map() → ndarray[Any, dtype[int64]]

Generate process lookup map for subdomains.

Map values:: -2: Wall Boundary Condition -1: No Assumption Boundary Condition >=0: proc_ID

Returns:: Map array with process IDs and boundary flags.
Return type:: np.ndarray

get_neighbor_ranks(sd_index: tuple[int, ...]) → dict[tuple[int, ...], int]

Determine the neighbor process rank for each feature.

Parameters:: sd_index (tuple[int, int, int]) – Index of the subdomain.
Returns:: Mapping from feature index to neighbor rank.
Return type:: dict

pmmoto.core.domain_discretization module

domain_discretization.py

Defines the DiscretizedDomain class for discretizing a physical domain into voxels.

class pmmoto.core.domain_discretization.DiscretizedDomain(voxels: tuple[int, ...] = (1, 1, 1), **kwargs: Any)

Bases: Domain

Discretize a physical domain into a voxel grid.

This class extends the base Domain class to include voxelization and calculation of voxel resolution and coordinates.

Initialize a discretized domain.

Parameters:

voxels (tuple[int, int, int], optional) – Number of voxels in each dimension.
**kwargs – Additional arguments passed to the base Domain class.

classmethod from_domain(domain: Domain, voxels: tuple[int, ...]) → Self

Create a DiscretizedDomain from an existing Domain and voxel counts.

Parameters:

domain (Domain) – The base domain object.
voxels (tuple[int, int, int]) – Number of voxels in each dimension.

Returns:

New discretized domain instance.

Return type:

DiscretizedDomain

static get_coords(box: tuple[tuple[float, float], ...], voxels: tuple[int, ...], resolution: tuple[float, ...]) → list[ndarray[Any, dtype[float64]]]

Determine the physical locations of voxel centroids.

Parameters:

box (tuple[tuple[float, float], ...]) – Physical bounds for each dimension.
voxels (tuple[int,int, int]) – Number of voxels in each dimension.
resolution (tuple[float, float, float]) – Voxel size in each dimension.

Returns:

List of arrays with centroid coordinates.

Return type:

list[np.ndarray]

get_resolution() → tuple[float, ...]

Calculate the physical size of each voxel in every dimension.

Returns:: tuple[float, float, float] Resolution (voxel size) in each dimension.

pmmoto.core.features module

features.py

This holds features of a cube: faces edges, and corners

class pmmoto.core.features.Corner(feature_id: tuple[int, ...], neighbor_rank: int, boundary_type: BoundaryType, global_boundary: bool | None = None)

Bases: Feature

Corner information for a subdomain.

Need to distinguish between internal and external corners. There are 8 external corners. All others are termed internal.

Initialize a Corner feature.

Parameters:

feature_id – Feature identifier (tuple).
neighbor_rank – Neighboring process rank.
boundary_type – Boundary type.
global_boundary – Whether this is a global boundary (bool).

get_periodic_correction() → tuple[int, ...]: Determine spatial correction factor (shift) if periodic

class pmmoto.core.features.Edge(feature_id: tuple[int, ...], neighbor_rank: int, boundary_type: BoundaryType, global_boundary: bool | None = None)

Bases: Feature

Edge information for a subdomain.

Need to distinguish between internal and external edges. There are 12 external corners. All others are termed internal.

Initialize an Edge feature.

Parameters:

feature_id – Feature identifier (tuple).
neighbor_rank – Neighboring process rank.
boundary_type – Boundary type.
global_boundary – Whether this is a global boundary (bool).

get_periodic_correction() → tuple[int, ...]: Determine spatial correction factor if periodic

class pmmoto.core.features.Face(feature_id: tuple[int, ...], neighbor_rank: int, boundary_type: BoundaryType, global_boundary: bool | None = None, inlet: bool | None = None, outlet: bool | None = None)

Bases: Feature

Face information for a subdomain.

Initialize a Face feature.

Parameters:

feature_id – Feature identifier (tuple).
neighbor_rank – Neighboring process rank.
boundary_type – Boundary type.
global_boundary – Whether this is a global boundary (bool).
inlet – Whether this face is an inlet (bool).
outlet – Whether this face is an outlet (bool).

get_direction() → bool

Determine if the face is point in the forward direction.

-1 for the non-negative feature id

Returns:: _description_
Return type:: _type_

get_periodic_correction() → tuple[int, ...]

Determine spatial correction factor if periodic.

Returns:: A tuple of correction values for each spatial dimension.

get_slice() → tuple[int | slice, ...]

Extract a 2D slice index for accessing a boundary face in a 3D array.

The method inspects self.feature_id, which defines the orientation of a face in 3D space. It returns a tuple with one fixed index (either 0 or -1) to select a specific boundary face, and slice(None) for the remaining dimensions.

Returns:: A 3-element tuple used to index a 2D face from a 3D array.
Return type:: Tuple[int | slice, … ]

get_strides(strides: tuple[int, ...]) → int

Return the stride corresponding to the active dimension in the feature ID.

This method is used to identify which stride (among the three dimensions) is relevant based on the nonzero entry in self.feature_id.

Parameters:: strides (Tuple[int, int, int]) – A 3-element tuple representing the strides of a NumPy array, typically from .strides.
Returns:: The stride value corresponding to the active axis.
Return type:: int

map_to_index() → int: Return the 1d-index for face.

class pmmoto.core.features.Feature(dim: int, feature_id: tuple[int, ...], neighbor_rank: int, boundary_type: BoundaryType, global_boundary: bool | None = None)

Bases: object

Base class for holding feature: {face, edge, corner} information.

This is the main abstraction for handling boundary conditions and parallel communication.

Initialize a Feature.

Parameters:

dim – dimension
feature_id – Feature identifier (tuple).
neighbor_rank – Neighboring process rank.
boundary_type – Boundary type (e.g., “wall”, “periodic”).
global_boundary – Whether this is a global boundary (bool).

compute_lower_face(lower_pad: int) → list[int]: Determine lower face voxels

compute_lower_neighbor(lower_pad: int) → list[int]: Determine lower neighbor face voxels

compute_upper_face(length: int, upper_pad: int) → list[int]: Determine upper face voxels

compute_upper_neighbor(length: int, upper_pad: int) → list[int]: Determine upper neighbor face voxels

convert_feature_id(index: tuple[int, ...] | None = None) → int

Convert the feature type to id.

Returns:: feature id
Return type:: int

get_voxels(voxels: tuple[int, ...], pad: tuple[tuple[int, int], ...] = ((0, 0), (0, 0), (0, 0))) → None: Determine the voxel indices for the feature

is_periodic(boundary_type: BoundaryType) → bool

Determine if a feature is periodic

Returns:: True if periodic
Return type:: bool

pmmoto.core.logging module

logging.py

Setup of the logger

class pmmoto.core.logging.MPIFormatter(*args: Any, **kwargs: Any)

Bases: Formatter

Custom formatter that includes MPI rank

Initialize the formatter with specified format strings.

Initialize the formatter either with the specified format string, or a default as described above. Allow for specialized date formatting with the optional datefmt argument. If datefmt is omitted, you get an ISO8601-like (or RFC 3339-like) format.

Use a style parameter of ‘%’, ‘{’ or ‘$’ to specify that you want to use one of %-formatting, str.format() ({}) formatting or string.Template formatting in your format string.

Changed in version 3.2: Added the style parameter.

format(record: LogRecord) → str: Add mpi information

pmmoto.core.logging.get_logger() → Logger: Get or create the logger instance

pmmoto.core.logging.setup_logger(name: str = 'pmmoto', log_dir: str = 'logs') → Logger

Configure logging for both serial and parallel runs

Parameters:

name – Logger name (default: “pmmoto”)
log_dir – Directory to store log files (default: “logs”)

Returns:

Configured logger instance

Return type:

logging.Logger

pmmoto.core.orientation module

orientation.py

Defines orientation and feature indexing utilities for PMMoTo subdomains.

pmmoto.core.orientation.get_boundary_id(boundary_index: tuple[int, ...]) → int

Determine boundary ID from a boundary index.

Parameters:

boundary_index (list[int] or tuple[int]) – Boundary index for [x, y, z],
-1 (values)
0
1.

Returns:

Encoded boundary ID.

Return type:

int

pmmoto.core.pmmoto module

pmmoto.py

pmmoto.core.pmmoto.initialize(voxels: tuple[int, ...], box: tuple[tuple[float, float], ...] = ((0, 1.0), (0, 1.0), (0, 1)), subdomains: tuple[int, ...] = (1, 1, 1), boundary_types: tuple[tuple[BoundaryType, BoundaryType], ...] = ((BoundaryType.END, BoundaryType.END), (BoundaryType.END, BoundaryType.END), (BoundaryType.END, BoundaryType.END)), inlet: tuple[tuple[bool, bool], ...] = ((False, False), (False, False), (False, False)), outlet: tuple[tuple[bool, bool], ...] = ((False, False), (False, False), (False, False)), reservoir_voxels: int = 0, rank: int = 0, pad: tuple[int, ...] = (1, 1, 1), verlet_domains: tuple[int, ...] = (1, 1, 1), return_subdomain: bool = False) → Subdomain | PaddedSubdomain | VerletSubdomain: Initialize PMMoTo domain and subdomain classes and check for valid inputs.

pmmoto.core.subdomain module

subdomains.py

Defines the Subdomain class for domain decomposition and parallelization in PMMoTo.

class pmmoto.core.subdomain.Subdomain(rank: int, decomposed_domain: DecomposedDomain)

Bases: DiscretizedDomain

Decompose the domain into subdomains for parallelization.

Initialize a Subdomain.

Parameters:

rank (int) – Rank of the subdomain.
decomposed_domain – Decomposed domain object.

check_boundary_type(type: BoundaryType) → bool: Determine if boundary type on subdomain.

get_boundary_types() → tuple[tuple[BoundaryType, BoundaryType], ...]

Determine the boundary type for each subdomain.

Returns:: Tuple specifying boundary type for each face
Return type:: tuple[tuple[str,str],…]

get_box(voxels: tuple[int, ...]) → tuple[tuple[float, float], ...]

Determine the bounding box for each subdomain.

Note

Subdomains are divided such that voxel spacing is constant.

Parameters:: voxels (tuple) – Number of voxels in each dimension.
Returns:: Bounding box for each dimension.
Return type:: tuple

get_centroid() → ndarray[Any, dtype[float64]]

Determine the centroid of a subdomain.

Returns:: Centroid coordinates.
Return type:: np.ndarray

get_global_boundary() → tuple[tuple[bool, bool], ...]

Determine if the subdomain is on the domain boundary.

For a subdomain to be on a domain boundary, the subdomain index must contain either 0 or the number of subdomains.

Returns:: Tuple of bools whether face is on boundary
Return type:: tuple[tuple[bool,bool],…]

get_img_index(coordinates: tuple[float, ...]) → tuple[int, ...] | None

Given coordinates, return the corresponding index in the img array.

Parameters:: coordinates (tuple[float, float, float]) – The (x, y, z) coordinates.
Returns:: The (i, j, k) index in the img array, or None if out of bounds.
Return type:: tuple[int, int, int] or None

static get_index(rank: int, subdomains: tuple[int, ...]) → tuple[int, ...]

Determine the index of the subdomain in the decomposition.

Parameters:

rank (int) – The rank of the subdomain.
subdomains (tuple[int, int, int]) – Number of subdomains in each dimension.

Returns:

Index of the subdomain in the decomposition grid.

Return type:

tuple[int, int, int]

get_inlet() → tuple[tuple[bool, bool], ...]

Determine if the subdomain lies on the inlet boundaries.

A subdomain is on an inlet if: - It is at the edge of the global domain (index is 0 or max in a dimension). - The corresponding global boundary type is 0 (denoting an inlet).

Returns:: bool if subdomain face is on inlet.
Return type:: tuple[tuple[bool, bool], …]

get_length() → tuple[float, ...]

Calculate the length of the subdomain in each dimension.

Returns:: Length in each dimension.
Return type:: tuple[float, …]

get_origin() → tuple[float, ...]

Determine the domain origin from box.

Returns:: Domain origin.
Return type:: tuple[float, …]

get_outlet() → tuple[tuple[bool, bool], ...]

Determine if the subdomain lies on the inlet boundaries.

A subdomain is on an outlet if: - It is at the edge of the global domain (index is 0 or max in a dimension). - The corresponding global boundary type is 0 (denoting an outlet).

Returns:: bool if subdomain face is on outlet.
Return type:: tuple[tuple[bool, bool], …]

get_own_voxels() → ndarray[Any, dtype[integer[Any]]]

Determine the index for the voxels owned by this subdomain.

Returns:: Array of indices for owned voxels.
Return type:: np.ndarray

get_radius() → Any

Determine the radius of a subdomain.

Returns:: Radius of the subdomain.
Return type:: float

static get_start(index: tuple[int, ...], domain_voxels: tuple[float, ...], subdomains: tuple[int, ...]) → tuple[int, ...]

Determine the start of the subdomain.

Parameters:

index (tuple[int, int, int]) – Subdomain index.
domain_voxels (tuple[float, float, float]) – Number of voxels per dimension
subdomains (tuple[int, int, int]) – Number of subdomains per dimension.

Returns:

Start voxel indices - minimum voxel ID.

Return type:

tuple[int, int, int]

static get_voxels(index: tuple[int, ...], domain_voxels: tuple[int, ...], subdomains: tuple[int, ...]) → tuple[int, ...]

Calculate number of voxels in each subdomain.

Parameters:

index (tuple) – Index of the subdomain.
domain_voxels (tuple) – Number of voxels in the full domain.
subdomains (tuple) – Number of subdomains in each dimension.

Returns:

tuple[int, int, int] Number of voxels in each dimension for this subdomain.

periodic_check() → bool

Determine if subdomain is periodic

Returns: bool if and feature is periodic

set_wall_bcs(img: ndarray[Any, dtype[T]]) → ndarray[Any, dtype[T]]

Force solid on external boundaries if wall boundary conditions are specified.

Parameters:: img (np.ndarray) – Image array to update.
Returns:: Updated image array with wall boundaries set to solid.
Return type:: np.ndarray

pmmoto.core.subdomain_features module

subdomain_features.py

Defines feature classes and utilities for handling subdomain faces, edges, and corners, including boundary and periodicity information.

class pmmoto.core.subdomain_features.SubdomainFeatures(subdomain: Subdomain | PaddedSubdomain, voxels: tuple[int, ...], pad: tuple[tuple[int, int], ...] = ((0, 0), (0, 0), (0, 0)))

Bases: object

Container for all features (faces, edges, corners) of a subdomain.

property all_features: Iterator[tuple[tuple[int, ...], Face | Edge | Corner]]: Iterates through features

collect_features() → None: Collect information for faces, edges, and corners for a subdomain.

collect_periodic_corrections() → dict[tuple[int, ...], tuple[int, ...]]

Collect periodic correction vectors for all periodic features.

Parameters:: features (dict) – Dictionary with keys “faces”, “edges”, “corners”.
Returns:: Mapping from feature_id to periodic correction tuple.
Return type:: dict

collect_periodic_features() → list[tuple[int, ...]]

Collect all periodic features from a features dictionary.

Returns:: List of feature_ids that are periodic.
Return type:: list

get_boundary_type(feature_id: tuple[int, ...], neighbor_rank: int) → BoundaryType

Determine the boundary type for a feature.

For a feature to be on a domain boundary, the subdomain index must contain either 0 or the number of subdomains.

Returns:: If feature is on domain boundary
Return type:: bool

get_feature_member(feature_id: tuple[int, ...], member_name: str) → Any

Get the value of a member from the Face, Edge, or Corner with feature_id.

Parameters:

feature_id (tuple[int, ...]) – The feature ID.
member_name (str) – Name of the member to retrieve.

Returns:

Value of the requested member.

Return type:

Any

Raises:

KeyError – If feature_id is not found.
AttributeError – If member_name does not exist.

get_features() → dict[tuple[int, ...], Face | Edge | Corner]: Return a dict of all features

get_global_boundary(feature_id: tuple[int, ...]) → bool

Determine if a feature is on the domain boundary.

For a feature to be on a domain boundary, the subdomain index must contain either 0 or the number of subdomains.

Returns:: If feature is on domain boundary
Return type:: bool

set_feature_voxels(voxels: tuple[int, ...]) → None: Set the own and neighbor voxels for each feature

pmmoto.core.subdomain_padded module

subdomain_padded.py

Defines the PaddedSubdomain class for handling subdomains with padding, facilitating parallel algorithms in PMMoTo.

class pmmoto.core.subdomain_padded.PaddedSubdomain(rank: int, decomposed_domain: DecomposedDomain, pad: tuple[int, ...] = (0, 0, 0), reservoir_voxels: int = 0)

Bases: Subdomain

Padded subdomain to facilitate development of parallel algorithms.

Initialize a PaddedSubdomain.

Parameters:

rank (int) – Rank of the subdomain.
decomposed_domain – Decomposed domain object.
pad (tuple[int, int, int], optional) – Padding for each dimension.
reservoir_voxels (int, optional) – Reservoir voxels to pad at inlet/outlet.

extend_padding(pad: tuple[int, ...]) → tuple[tuple[tuple[int, int], ...], SubdomainFeatures]

Extend pad to boundaries of subdomain.

Padding is applied to all boundaries except ‘end’ and ‘wall’ boundary types, where pad is limited to 1. Padding must be equal on opposite features.

Parameters:: pad (tuple[int, ...]) – Pad length for each dimension.
Returns:: (pad, extended_features)
Return type:: tuple

get_own_voxels() → ndarray[Any, dtype[int64]]

Determine the index for the voxels owned by this subdomain.

Returns:: Array of indices for owned voxels.
Return type:: np.ndarray

get_padded_box() → tuple[tuple[float, float], ...]

Determine the bounding box for each subdomain.

Note

Subdomains are divided such that voxel spacing is constant.

Returns:: Bounding box for each dimension.
Return type:: tuple

static get_padded_start(index: tuple[int, ...], domain_voxels: tuple[int, ...], subdomains: tuple[int, ...], pad: tuple[tuple[int, int], ...] = ((0, 0), (0, 0), (0, 0)), reservoir_pad: tuple[tuple[int, int], ...] = ((0, 0), (0, 0), (0, 0))) → tuple[int, ...]

Determine the start of the subdomain. used for saving as vtk.

Start is the minimum voxel ID :param index: subdomain index :type index: tuple[int, int, int]

Returns:: start
Return type:: tuple[int,…]

get_padded_voxels() → tuple[int, ...]

Calculate number of voxels in each subdomain.

This includes padding and reservoir.

Returns:: Number of voxels in each dimension.
Return type:: tuple[int, …]

get_padding(pad: tuple[int, ...]) → tuple[tuple[int, int], ...]

Compute the padding for each dimension of the subdomain.

Padding is applied to all boundaries except ‘end’ and ‘wall’ boundary types, where pad is limited to 1. Padding must be equal on opposite features.

Parameters:: pad (tuple[int, ...]) – Pad length for each dimension.
Returns:: Padding for each dimension.
Return type:: tuple[tuple[int, int], …]

get_reservoir_padding(reservoir_voxels: int) → tuple[tuple[int, int], ...]

Determine inlet/outlet info and pad image (only inlet).

Parameters:: reservoir_voxels (int) – Number of reservoir voxels to pad.
Returns:: Reservoir padding for each dimension.
Return type:: tuple[tuple[int, int], …]

update_reservoir(img: ndarray[Any, dtype[T]], value: T) → ndarray[Any, dtype[T]]

Enforce a constant value in reservoir regions.

Parameters:

img (np.ndarray) – Image array to update.
value – Value to set in reservoir regions.

Returns:

Updated image array.

Return type:

np.ndarray

pmmoto.core.subdomain_verlet module

subdomain_verlet.py

class pmmoto.core.subdomain_verlet.VerletSubdomain(rank: int, decomposed_domain: DecomposedDomain, verlet_domains: tuple[int, ...], pad: tuple[int, ...] = (0, 0, 0), reservoir_voxels: int = 0)

Bases: PaddedSubdomain

Verlet subdomains divide a subdomain into smaller Verlet domains.

Initialize a PaddedSubdomain.

Parameters:

rank (int) – Rank of the subdomain.
decomposed_domain – Decomposed domain object.
pad (tuple[int, int, int], optional) – Padding for each dimension.
reservoir_voxels (int, optional) – Reservoir voxels to pad at inlet/outlet.

get_maximum_diameter() → ndarray[Any, dtype[float64]]: Determine the maximum diameter of the verlet subdomain

get_verlet_box() → dict[int, tuple[tuple[int, int], ...]]: Determine the box of the verlet subdomain

get_verlet_centroid() → ndarray[Any, dtype[float64]]: Determine the center of the verlet domain

get_verlet_loop() → ndarray[Any, dtype[uint64]]: Collect the loop information for each verlet subdomain

get_verlet_voxels() → tuple[tuple[int, ...], ...]: Determine the number of voxels for each verlet subdomain

pmmoto.core.utils module

Core utility functions for PMMoTo.

This module provides utility functions for array manipulation, validation, MPI-aware operations, and subdomain/grid management.

pmmoto.core.utils.bin_image(subdomain: Subdomain, img: NDArray[T], own: bool = True) → dict[T, int]

Count the number of times each unique element occurs in the input array.

Parameters:

subdomain – Subdomain object.
img (np.ndarray) – Input array.
own (bool) – If True, use only owned voxels.

Returns:

Mapping from element value to count.

Return type:

dict

pmmoto.core.utils.constant_pad_img(img: ndarray[Any, dtype[T]], pad: tuple[tuple[int, int], ...], pad_value: int | float) → ndarray[Any, dtype[T]]

Pad a grid with a constant value.

Parameters:

img (np.ndarray) – Input array.
pad (list or tuple) – Padding amounts for each dimension.
pad_value (scalar) – Value to use for padding.

Returns:

The padded array.

Return type:

np.ndarray

pmmoto.core.utils.determine_maximum(img: ndarray[Any, dtype[T]]) → int | float

Determine the global maximum of an input image.

Parameters:: img (np.ndarray) – Input array.
Returns:: Global maximum value.
Return type:: scalar

pmmoto.core.utils.own_img(subdomain: Subdomain, img: NDArray[T], own_voxels: None | NDArray[np.integer[Any]] = None) → NDArray[T]

Return array with only nodes owned by the current process.

Parameters:

subdomain – Subdomain object with get_own_voxels method.
img (np.ndarray) – Input image.
own_voxels – NDArray of size 2*dims with bounds of image to extract

Returns:

Array of owned voxels.

Return type:

np.ndarray

pmmoto.core.utils.phase_exists(img: ndarray[Any, dtype[T]], phase: int | float) → bool

Determine if a phase exists in the grid (globally).

Parameters:

img (np.ndarray) – Input array.
phase (int) – Phase value to check.

Returns:

True if phase exists, False otherwise.

Return type:

bool

pmmoto.core.utils.unpad(img: ndarray[Any, dtype[T]], pad: tuple[tuple[int, int], ...]) → ndarray[Any, dtype[T]]

Remove padding from a NumPy array.

Parameters:

img (np.ndarray) – The padded array.
pad (list or tuple) – Padding

Returns:

The unpadded array.

Return type:

np.ndarray

pmmoto.core.voxels module

voxel.py

Core voxel operations for PMMoTo.

pmmoto.core.voxels.count_label_voxels(img: ndarray[Any, dtype[T]], map: dict[int, int]) → dict[int, int]

Count the number of voxels for each label in the grid.

Parameters:

img – Grid array.
map – Mapping array.

Returns:

None

pmmoto.core.voxels.gen_img_to_label_map(img: ndarray[Any, dtype[INT]], labels: ndarray[Any, dtype[INT2]]) → dict[INT2, INT]

Generate a mapping from grid indices to labels.

Parameters:

img – Grid array.
labels – Label array.

Returns:

Mapping array.

pmmoto.core.voxels.gen_inlet_label_map(subdomain: Subdomain | PaddedSubdomain | VerletSubdomain, label_img: ndarray[Any, dtype[T]]) → ndarray[Any, dtype[T]]

Determine which face is on inlet.

Currently restricted to a single face.

Parameters:

subdomain – Subdomain object.
label_img – Labeled image.

Returns:

Array of inlet labels.

Return type:

np.ndarray

pmmoto.core.voxels.gen_outlet_label_map(subdomain: Subdomain | PaddedSubdomain | VerletSubdomain, label_img: ndarray[Any, dtype[T]]) → ndarray[Any, dtype[T]]

Determine which face is on outlet.

Currently restricted to a single face.

Parameters:

subdomain – Subdomain object.
label_img – Labeled image.

Returns:

Array of outlet labels.

Return type:

np.ndarray

pmmoto.core.voxels.get_boundary_voxels(subdomain: Subdomain | PaddedSubdomain | VerletSubdomain, img: ndarray[Any, dtype[INT]], neighbors_only: bool = False) → dict[tuple[int, ...], dict[str, ndarray[Any, dtype[INT]]]]

Return the values on the boundary features.

The features are divided into: - own: feature voxels owned by subdomain - neighbor: feature voxels owned by a neighbor subdomain

Parameters:

subdomain – Subdomain object.
img – Image array.
neighbors_only (bool) – If True, only include neighbor voxels.

Returns:

Dictionary of boundary voxels.

Return type:

dict

pmmoto.core.voxels.get_id(index: tuple[int, ...], total_voxels: tuple[int, ...]) → uint64

Get the global or local ID for a voxel.

Parameters:

index – 3D index of the voxel (x, y, z).
total_voxels – Number of voxels in each dimension.

Returns:

Global or local ID of the voxel.

Return type:

int

Note

Periodic boundary conditions are applied by using modulo arithmetic.

pmmoto.core.voxels.get_nearest_boundary_index(subdomain: Subdomain | PaddedSubdomain | VerletSubdomain, img: ndarray[Any, dtype[T]], label: int, dimension: None | int = None, which_voxels: Literal['all', 'own', 'pad'] = 'all') → dict[tuple[int, ...], ndarray[Any, dtype[T]]]

Determine the nearest boundary index to a given label

If which_voxels == “all”, always use base version. If which_voxels is “own”/ “pad”, use extended version.

pmmoto.core.voxels.match_global_boundary_voxels(matches: dict[tuple[int, INT], dict[str, tuple[int, INT]]], label_count: int) → tuple[dict[int, dict[int, INT]], int]

Generate a global label map for matched boundary voxels.

Parameters:

subdomain – Subdomain object.
matches – Matches dictionary.
label_count – Number of labels on this rank.

Returns:

(final_map, global_label_count)

Return type:

tuple

pmmoto.core.voxels.match_neighbor_boundary_voxels(subdomain: Subdomain | PaddedSubdomain | VerletSubdomain, boundary_voxels: dict[tuple[int, ...], dict[str, ndarray[Any, dtype[INT]]]], recv_data: dict[tuple[int, ...], dict[str, ndarray[Any, dtype[INT]]]], skip_zero: bool = False) → dict[tuple[int, INT], dict[str, tuple[int, INT]]]

Match boundary voxels with subdomain neighbor voxels and return unique matches.

Parameters:

subdomain – Subdomain object containing feature information.
boundary_voxels – Dictionary with ‘own’ and ‘neighbor’ boundary voxel data.
recv_data – Received data containing neighbor voxel information.
skip_zero (bool) – If True, skip matches with label 0.

Returns:

Unique matches in the format:: key: (subdomain rank, own voxel) neighbor: list[neighbor rank, neighbor voxel)]

Return type:

dict

pmmoto.core.voxels.renumber_image(img: ndarray[Any, dtype[INT2]], conversion_map: dict[INT2, INT]) → ndarray[Any, dtype[INT]]

Renumber an image using a provided mapping.

Parameters:

img (Any) – The image to be renumbered.
conversion_map (dict) – A dictionary mapping current image IDs to new image IDs. Example: {1: 101, 2: 102, …} Note: All IDs in img must be defined in conversion_map.

Returns:

The renumbered image, with IDs replaced based on the mapping.

Return type:

Any

Note

This function assumes all required IDs are present in the conversion_map. No error handling is performed for missing or invalid keys.

Example

img = [[1, 2], [2, 1]] conversion_map = {1: 101, 2: 102} renumber_image(img, conversion_map) # Output: [[101, 102], [102, 101]]