Source code for physicsnemo.mesh.domain_mesh

# SPDX-FileCopyrightText: Copyright (c) 2023 - 2026 NVIDIA CORPORATION & AFFILIATES.
# SPDX-FileCopyrightText: All rights reserved.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# ``tensorclass`` adds a class-scoped ``float`` method. Qualify scalar
# annotations that must remain resolvable under Python's deferred lookup.
import builtins
from collections.abc import Callable, Iterator, Sequence
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal, Self

import torch
from jaxtyping import Bool, Float
from tensordict import TensorDict, tensorclass

from physicsnemo.mesh.mesh import Mesh, _requested_float_dtype
from physicsnemo.mesh.transformations.deform.ffd import _FFDBasis
from physicsnemo.mesh.utilities.mesh_repr import format_mesh_repr

if TYPE_CHECKING:
    import matplotlib.axes
    import pyvista


[docs] @tensorclass class DomainMesh: r"""A simulation domain represented as an interior mesh with named boundary meshes. A ``DomainMesh`` groups an interior :class:`Mesh` (either a volumetric mesh with full connectivity or a point cloud) together with zero or more boundary :class:`Mesh` objects keyed by boundary condition type (e.g. ``"no_slip"``, ``"inlet"``, ``"farfield"``), plus optional domain-level metadata in ``global_data``. ``DomainMesh`` exposes sparse world-space :meth:`morph`, global :meth:`radial_basis_function_deform`, and lattice :meth:`free_form_deform`, but no dense ``displace``. Component point counts and fields can differ. One shared control field transfers consistently across every component. The semantic contract is that the boundary meshes, if merged, form a watertight enclosure around the interior mesh. This is documented but not enforced at construction time; call :meth:`is_boundary_watertight` to verify explicitly. Because ``DomainMesh`` is a tensorclass, standard TensorDict operations like :meth:`to`, :meth:`clone`, and :meth:`pin_memory` propagate to ``interior``, all ``boundaries``, and ``global_data`` automatically. Parameters ---------- interior : Mesh The interior region mesh. Can be a volumetric mesh with full simplicial connectivity (triangles, tetrahedra) or a bare point cloud. boundaries : dict[str, Mesh] or TensorDict[str, Mesh], optional Boundary condition meshes keyed by BC type name. If a ``dict`` is provided, it is automatically converted to a :class:`TensorDict`. Defaults to an empty collection. global_data : dict[str, torch.Tensor] or TensorDict, optional Domain-level quantities that apply to the entire simulation (e.g. Reynolds number, angle of attack, Mach number). If a ``dict`` is provided, it is automatically converted to a :class:`TensorDict`. Defaults to an empty collection. Raises ------ TypeError If ``interior`` is not a :class:`Mesh`, or if any value in ``boundaries`` is not a :class:`Mesh`. ValueError If any boundary mesh has a different ``n_spatial_dims`` than ``interior``. Examples -------- Create a domain with a volumetric interior and two boundary patches: >>> import torch >>> from physicsnemo.mesh import Mesh, DomainMesh >>> interior = Mesh(points=torch.randn(100, 3)) >>> wall = Mesh( ... points=torch.tensor([[0., 0., 0.], [1., 0., 0.], [0., 1., 0.]]), ... cells=torch.tensor([[0, 1, 2]]), ... ) >>> inlet = Mesh( ... points=torch.tensor([[2., 0., 0.], [3., 0., 0.], [2., 1., 0.]]), ... cells=torch.tensor([[0, 1, 2]]), ... ) >>> dm = DomainMesh( ... interior=interior, ... boundaries={"no_slip": wall, "inlet": inlet}, ... global_data={"Re": torch.tensor(1e6), "AoA": torch.tensor(5.0)}, ... ) >>> dm.n_boundaries 2 >>> dm.boundary_names ['inlet', 'no_slip'] Create a domain with no boundaries (e.g. a standalone point cloud): >>> dm = DomainMesh(interior=Mesh(points=torch.randn(50, 3))) >>> dm.n_boundaries 0 Move everything to GPU: >>> dm_gpu = dm.to("cuda") # doctest: +SKIP """ interior: Mesh boundaries: TensorDict[str, Mesh] global_data: TensorDict def __init__( self, interior: Mesh, boundaries: dict[str, Mesh] | TensorDict | None = None, global_data: dict[str, torch.Tensor] | TensorDict | None = None, ) -> None: self.interior = interior self.boundaries = boundaries # normalized by __post_init__ self.global_data = global_data # normalized by __post_init__ # tensorclass only auto-calls __post_init__ from the *generated* __init__ # (same semantics as dataclasses). Since we define a custom __init__, # we must call it explicitly. During load(), tensorclass calls it # automatically, so __post_init__ is the single source of truth for # defaults, coercions, and validation. self.__post_init__() def __post_init__(self) -> None: """Normalize fields and validate invariants. Called automatically during ``load()`` by tensorclass, and explicitly from ``__init__`` during normal construction. This is the single source of truth for all default values, type coercions, and shape validation. """ ### boundaries: coerce dict -> TensorDict, None -> empty TensorDict if isinstance(self.boundaries, dict): self.boundaries = TensorDict(self.boundaries, batch_size=[]) elif self.boundaries is None: self.boundaries = TensorDict({}, batch_size=[]) else: self.boundaries.batch_size = torch.Size([]) ### global_data: coerce dict -> TensorDict, None -> empty TensorDict if isinstance(self.global_data, TensorDict): self.global_data.batch_size = torch.Size([]) else: self.global_data = TensorDict( {} if self.global_data is None else dict(self.global_data), batch_size=torch.Size([]), ) ### Validate types and dimensional consistency if not torch.compiler.is_compiling(): if not isinstance(self.interior, Mesh): raise TypeError( f"`interior` must be a Mesh, got {type(self.interior).__name__}." ) expected_spatial_dims = self.interior.n_spatial_dims for name in self.boundaries.keys(): bc_mesh = self.boundaries[name] if not isinstance(bc_mesh, Mesh): raise TypeError( f"All boundary values must be Mesh instances, but " f"boundaries[{name!r}] is {type(bc_mesh).__name__}." ) if bc_mesh.n_spatial_dims != expected_spatial_dims: raise ValueError( f"All meshes must share the same spatial dimension " f"({expected_spatial_dims}), but boundaries[{name!r}] " f"has n_spatial_dims={bc_mesh.n_spatial_dims}." )
[docs] def apply_to_meshes( self, fn: Callable[[Mesh], Mesh], *, interior: bool = True, boundaries: bool = True, ) -> "DomainMesh": r"""Apply a Mesh-to-Mesh function to meshes in the domain. By default, ``fn`` is called on the ``interior`` and on each boundary mesh. Use the keyword flags to apply selectively. Components that are skipped are cloned unchanged. Domain-level ``global_data`` is always cloned unchanged. All built-in operations (``translate``, ``rotate``, ``subdivide``, ``clean``, etc.) delegate here. This is distinct from the inherited tensorclass :meth:`apply`, which recursively maps a ``Tensor -> Tensor`` callable across every leaf tensor. Use :meth:`apply` for tensor-level transforms (e.g. dtype casting) and :meth:`apply_to_meshes` for mesh-level transforms. Parameters ---------- fn : Callable[[Mesh], Mesh] A function that takes a :class:`Mesh` and returns a :class:`Mesh`. interior : bool If ``True`` (default), apply ``fn`` to the interior mesh. boundaries : bool If ``True`` (default), apply ``fn`` to every boundary mesh. Returns ------- DomainMesh New domain with the transformed meshes. Examples -------- Convert every mesh to a point cloud (drop connectivity): >>> dm_cloud = dm.apply_to_meshes(lambda m: Mesh(points=m.points)) # doctest: +SKIP Subdivide only the boundaries (e.g. to match a finer interior): >>> dm2 = dm.apply_to_meshes( # doctest: +SKIP ... lambda m: m.subdivide(levels=1), boundaries=True, interior=False ... ) """ return DomainMesh( interior=fn(self.interior) if interior else self.interior.clone(), boundaries=( self.boundaries.apply(fn, call_on_nested=True) if boundaries else self.boundaries.clone() ), global_data=self.global_data.clone(), )
if TYPE_CHECKING: def to(self, *args: Any, **kwargs: Any) -> Self: """Move domain and all attached data to specified device/dtype. All tensors in ``interior``, every mesh in ``boundaries``, and ``global_data`` are moved together. Parameters ---------- *args : Any Positional arguments passed to the underlying tensorclass ``to`` method. Common usage: ``dm.to("cuda")`` or ``dm.to(torch.float32)``. **kwargs : Any Keyword arguments passed to the underlying tensorclass ``to`` method. Keyword Arguments ----------------- device : torch.device, optional The desired device. dtype : torch.dtype, optional The desired floating-point or complex dtype. non_blocking : bool, optional Whether the transfer should be non-blocking. Returns ------- DomainMesh A new DomainMesh on the target device/dtype, or the same instance if no changes were required. Examples -------- >>> dm_gpu = dm.to("cuda") # doctest: +SKIP >>> dm_cpu = dm.to(device="cpu") # doctest: +SKIP """ ... def clone(self) -> Self: """Return a deep clone of this DomainMesh. All tensors are copied (independent storage); the clone can be modified without affecting the original. """ ... def save( self, prefix: str | Path | None = None, copy_existing: bool = False, *, num_threads: int = 0, return_early: bool = False, share_non_tensor: bool = False, ) -> Self: """Save the domain mesh to disk as memory-mapped tensors. Writes ``interior``, all ``boundaries``, and ``global_data`` to a directory tree of ``.memmap`` files. Proxy for the tensorclass ``memmap()`` method. This is the recommended serialization method. Compared to ``torch.save`` (pickle-based), memmap serialization is faster (parallel I/O across files), safer (no arbitrary code execution on load), and supports partial loading. Parameters ---------- prefix : str, Path, or None Directory path where the memory-mapped files will be written. If ``None``, a temporary directory is used. copy_existing : bool If ``True``, copy tensors that are already memory-mapped to the new location. num_threads : int Number of threads for parallel I/O (0 = sequential). return_early : bool If ``True``, return before all data is flushed to disk. share_non_tensor : bool If ``True``, share non-tensor data across processes. Returns ------- DomainMesh A new DomainMesh backed by the on-disk memory-mapped storage. Examples -------- >>> dm.save("/path/to/domain_mesh") # doctest: +SKIP >>> reloaded = DomainMesh.load("/path/to/domain_mesh") # doctest: +SKIP """ ... @classmethod def load( cls, prefix: str | Path, device: torch.device | None = None, non_blocking: bool = False, ) -> Self: """Load a previously saved domain mesh from disk. Reads a directory tree of memory-mapped tensors written by :meth:`save` and reconstructs the ``DomainMesh`` instance, including the ``interior`` mesh, all ``boundaries``, and ``global_data``. Proxy for the tensorclass ``load_memmap()`` class method. Parameters ---------- prefix : str or Path Path to the directory created by :meth:`save`. device : torch.device or None If provided, move all tensors to this device after loading. non_blocking : bool Whether device transfers should be non-blocking. Returns ------- DomainMesh The reconstructed DomainMesh instance. Examples -------- >>> dm = DomainMesh.load("/path/to/domain_mesh") # doctest: +SKIP """ ... ### Geometric Transforms
[docs] def translate( self, offset: Float[torch.Tensor, " n_spatial_dims"] | Sequence[float], ) -> "DomainMesh": r"""Translate all meshes in the domain by a constant offset. Delegates to :meth:`Mesh.translate` for each mesh. Parameters ---------- offset : torch.Tensor or Sequence[float] Translation vector, shape :math:`(S,)` where :math:`S` is ``n_spatial_dims``. Returns ------- DomainMesh New domain with translated geometry. """ return self.apply_to_meshes(lambda m: m.translate(offset=offset))
[docs] def rotate( self, angle: float, axis: Float[torch.Tensor, " n_spatial_dims"] | Sequence[float] | Literal["x", "y", "z"] | None = None, center: Float[torch.Tensor, " n_spatial_dims"] | Sequence[float] | None = None, transform_point_data: bool | TensorDict = False, transform_cell_data: bool | TensorDict = False, transform_global_data: bool | TensorDict = False, ) -> "DomainMesh": r"""Rotate all meshes in the domain about an axis. Builds a rotation matrix and delegates to :meth:`transform`. Center handling uses translate-rotate-translate at the domain level, so domain-level :attr:`global_data` vectors are correctly rotated but not translated (vectors are translation-invariant). Parameters ---------- angle : float Rotation angle in radians. axis : torch.Tensor or Sequence[float] or {"x", "y", "z"}, optional Rotation axis vector, shape :math:`(D_s,)`. Use ``None`` for 2D. center : torch.Tensor or Sequence[float], optional Center point for rotation, shape :math:`(D_s,)`. transform_point_data : bool or TensorDict Controls transformation of ``point_data`` fields. ``True`` transforms all compatible fields; a ``TensorDict`` (or ``dict``) with scalar bool leaves selects specific fields. transform_cell_data : bool or TensorDict Same semantics, for ``cell_data``. transform_global_data : bool or TensorDict Same semantics, for each mesh's ``global_data`` and the domain-level :attr:`global_data`. Returns ------- DomainMesh New domain with rotated geometry. """ if center is not None: c = torch.as_tensor( center, device=self.interior.points.device, dtype=self.interior.points.dtype, ) return ( self.translate(-c) .rotate( angle=angle, axis=axis, center=None, transform_point_data=transform_point_data, transform_cell_data=transform_cell_data, transform_global_data=transform_global_data, ) .translate(c) ) from physicsnemo.mesh.transformations.geometric import rotation_matrix R = rotation_matrix( angle=angle, axis=axis, n_spatial_dims=self.interior.n_spatial_dims, device=self.interior.points.device, dtype=self.interior.points.dtype, ) return self.transform( matrix=R, transform_point_data=transform_point_data, transform_cell_data=transform_cell_data, transform_global_data=transform_global_data, assume_invertible=True, )
[docs] def scale( self, factor: float | Float[torch.Tensor, " n_spatial_dims"], center: Float[torch.Tensor, " n_spatial_dims"] | Sequence[float] | None = None, transform_point_data: bool | TensorDict = False, transform_cell_data: bool | TensorDict = False, transform_global_data: bool | TensorDict = False, assume_invertible: bool | None = None, ) -> "DomainMesh": r"""Scale all meshes in the domain by specified factor(s). Builds a scale matrix and delegates to :meth:`transform`. Center handling uses translate-scale-translate at the domain level. Parameters ---------- factor : float or torch.Tensor Scale factor (scalar) or per-dimension factors, shape :math:`(D_s,)`. center : torch.Tensor or Sequence[float], optional Center point for scaling, shape :math:`(D_s,)`. transform_point_data : bool or TensorDict Controls transformation of ``point_data`` fields. ``True`` transforms all compatible fields; a ``TensorDict`` (or ``dict``) with scalar bool leaves selects specific fields. transform_cell_data : bool or TensorDict Same semantics, for ``cell_data``. transform_global_data : bool or TensorDict Same semantics, for each mesh's ``global_data`` and the domain-level :attr:`global_data`. assume_invertible : bool or None, optional Controls cache propagation. See :meth:`Mesh.scale`. Returns ------- DomainMesh New domain with scaled geometry. """ if center is not None: c = torch.as_tensor( center, device=self.interior.points.device, dtype=self.interior.points.dtype, ) return ( self.translate(-c) .scale( factor=factor, center=None, transform_point_data=transform_point_data, transform_cell_data=transform_cell_data, transform_global_data=transform_global_data, assume_invertible=assume_invertible, ) .translate(c) ) from physicsnemo.mesh.transformations.geometric import scale_matrix M = scale_matrix( factor=factor, n_spatial_dims=self.interior.n_spatial_dims, device=self.interior.points.device, dtype=self.interior.points.dtype, ) return self.transform( matrix=M, transform_point_data=transform_point_data, transform_cell_data=transform_cell_data, transform_global_data=transform_global_data, assume_invertible=assume_invertible, )
[docs] def transform( self, matrix: Float[torch.Tensor, "new_n_spatial_dims n_spatial_dims"], transform_point_data: bool | TensorDict = False, transform_cell_data: bool | TensorDict = False, transform_global_data: bool | TensorDict = False, assume_invertible: bool | None = None, ) -> "DomainMesh": r"""Apply a linear transformation to all meshes in the domain. This is the single point of contact for domain-level :attr:`global_data` transformation. Both :meth:`rotate` and :meth:`scale` delegate here after building their matrix. Parameters ---------- matrix : torch.Tensor Transformation matrix, shape :math:`(S', S)`. transform_point_data : bool or TensorDict Controls transformation of ``point_data`` fields. ``True`` transforms all compatible fields; a ``TensorDict`` (or ``dict``) with scalar bool leaves selects specific fields. transform_cell_data : bool or TensorDict Same semantics, for ``cell_data``. transform_global_data : bool or TensorDict Same semantics, for each mesh's ``global_data`` and the domain-level :attr:`global_data`. assume_invertible : bool or None, optional Controls cache propagation. See :meth:`Mesh.transform`. Returns ------- DomainMesh New domain with transformed geometry. """ result = self.apply_to_meshes( lambda m: m.transform( matrix=matrix, transform_point_data=transform_point_data, transform_cell_data=transform_cell_data, transform_global_data=transform_global_data, assume_invertible=assume_invertible, ) ) if transform_global_data is not False: from physicsnemo.mesh.transformations.geometric import ( _normalize_transform_mask, _transform_tensordict, ) _transform_tensordict( result.global_data, matrix, self.interior.n_spatial_dims, "global_data", mask=_normalize_transform_mask(transform_global_data), ) return result
[docs] def morph( self, control_points: torch.Tensor, control_displacements: torch.Tensor, *, radius: builtins.float | torch.Tensor, point_weights: str | tuple[str, ...] | None = None, kernel: Literal["wendland_c2"] = "wendland_c2", implementation: Literal["torch", "warp"] | None = None, ) -> "DomainMesh": """Morph the interior and all boundaries with one world-space field. The same control coordinates, displacements, radii, and backend are used for every component, so coincident interior/boundary points receive the same motion when ``point_weights`` is ``None``. When supplied, ``point_weights`` is a common :attr:`Mesh.point_data` key (or nested tuple key) resolved on each component independently; raw point-weight tensors are intentionally rejected because component point counts differ. A common key does not require equal values: coincident component points remain coincident only when their resolved point weights also match. Parameters ---------- control_points : torch.Tensor World-coordinate controls with shape ``(n_controls, n_spatial_dims)`` and the same float32 or float64 dtype and device as every component's points. control_displacements : torch.Tensor Displacement vectors, not destination coordinates, with the same shape, dtype, and device as ``control_points``. radius : float or torch.Tensor Support distance in domain coordinate units. Supply a scalar or one radius per control. A tensor radius must match the control dtype and device; every value must remain positive and finite but is not validated at runtime. point_weights : str, tuple[str, ...], or None Optional point-data key present in every component and resolved independently on each component. Resolved tensors must have one common dtype; floating-point weights match the component point dtype. Raw tensors are not accepted. kernel : {"wendland_c2"}, optional Compact radial kernel used to blend control displacements. Default is ``"wendland_c2"``. implementation : {"torch", "warp"} or None Backend override. Auto dispatch uses Torch on CPU and Warp on CUDA when Warp is available, otherwise Torch. Returns ------- DomainMesh New domain with morphed component meshes and unchanged domain data. Notes ----- Connectivity and attached mesh and domain data are retained. Attached vector and tensor fields are treated as Lagrangian data and are not pushed forward. Geometry caches are invalidated and topology caches are retained on each component. Parameterize learned radii to remain positive, for example as ``torch.nn.functional.softplus(raw_radius) + eps``. Morphing does not automatically detect inverted, degenerate, or self-intersecting cells. Use each component mesh's :meth:`Mesh.validate` method explicitly when required. """ if not isinstance(control_points, torch.Tensor): raise TypeError( "control_points must be a torch.Tensor, got " f"{type(control_points).__name__}" ) if not isinstance(control_displacements, torch.Tensor): raise TypeError( "control_displacements must be a torch.Tensor, got " f"{type(control_displacements).__name__}" ) if point_weights is not None and not isinstance(point_weights, (str, tuple)): raise TypeError( "DomainMesh.morph point_weights must be a common point_data " "key/path, not a raw tensor" ) from physicsnemo.mesh.transformations.deform._utils import ( _resolve_domain_point_weights, ) components: list[tuple[str, Mesh]] = [("interior", self.interior)] components.extend( (f"boundaries[{name!r}]", self.boundaries[name]) for name in self.boundaries.keys() ) resolved_point_weights = _resolve_domain_point_weights( components, point_weights, control_points, "control_points" ) from physicsnemo.nn.functional.geometry.deform import morph_points def apply_field( combined_points: Float[torch.Tensor, "n_points n_spatial_dims"], combined_point_weights: Bool[torch.Tensor, " n_points"] | Float[torch.Tensor, " n_points"] | None, ) -> Float[torch.Tensor, "n_points n_spatial_dims"]: return morph_points( combined_points, control_points, control_displacements, radius=radius, point_weights=combined_point_weights, kernel=kernel, implementation=implementation, ) return self._deform_components(components, resolved_point_weights, apply_field)
[docs] def radial_basis_function_deform( self, control_points: Float[torch.Tensor, "n_controls n_spatial_dims"], control_displacements: Float[torch.Tensor, "n_controls n_spatial_dims"], *, kernel: Literal["thin_plate_spline"] = "thin_plate_spline", polynomial: builtins.bool = True, smoothing: builtins.float = 0.0, point_weights: str | tuple[str, ...] | None = None, implementation: Literal["torch", "warp"] | None = None, ) -> "DomainMesh": """Deform every component with one global thin-plate-spline RBF field. The same controls, fitted coefficients, kernel, and evaluation backend are shared across the interior and every boundary. With no point weights, coincident component points therefore receive identical motion. When supplied, ``point_weights`` is a common :attr:`Mesh.point_data` key (or nested tuple key) resolved independently on each component. Each resolved weight scales the fitted field at its point. Coincident component points therefore receive identical motion only when their resolved weights match. Raw weight tensors are rejected because component point counts differ. Parameters ---------- control_points : torch.Tensor World-coordinate controls with shape ``(n_controls, n_spatial_dims)`` and the same float32 or float64 dtype and device as every component's points. control_displacements : torch.Tensor Displacement vectors, not destination coordinates, with the same shape, dtype, and device as ``control_points``. kernel : {"thin_plate_spline"}, optional Radial kernel used by the interpolant. Default is ``"thin_plate_spline"``. polynomial : bool, optional Add the standard affine polynomial tail and side constraints. When controls are present, this requires at least ``D + 1`` distinct controls that span the ambient affine basis and form a nonsingular augmented system. Default is ``True``. smoothing : float, optional Nonnegative diagonal regularization. With a nonsingular control layout, zero interpolates the control displacements up to solver precision. Positive values relax interpolation accuracy. Default is ``0.0``. point_weights : str, tuple[str, ...], or None Optional point-data key present in every component. Resolved tensors must use one common bool or floating dtype. Floating weights must match component point dtypes. Every resolved tensor must be on the same device as its component's points. Raw tensors are not accepted. implementation : {"torch", "warp"} or None Field-evaluation backend. Both paths use PyTorch for the coefficient solve. Automatic dispatch uses Torch on CPU. On CUDA, it uses Warp when available and otherwise Torch. Returns ------- DomainMesh New domain with deformed component meshes and unchanged domain data. Raises ------ TypeError If control tensors or Python arguments have unsupported types, or if tensor dtypes are unsupported or mismatched. ValueError If component data, tensor shapes, devices, control layout, point weights, or RBF options are invalid. KeyError If a point-data key is missing or ``implementation`` does not name a registered backend. ImportError If an explicitly requested backend is unavailable. RuntimeError If runtime validation or coefficient fitting fails, including for a singular system or during CUDA Graph capture. Notes ----- The thin-plate-spline field has global support. Connectivity and attached mesh and domain data are retained. Attached vector and tensor fields are treated as Lagrangian data and are not pushed forward. Geometry caches are invalidated and topology caches are retained on each component. The operation does not detect inverted, degenerate, or self-intersecting cells. Use each component mesh's :meth:`Mesh.validate` method explicitly when required. Coefficient fitting is not supported inside CUDA Graph capture because the singular-system check requires host interaction. """ if not isinstance(control_points, torch.Tensor): raise TypeError( "control_points must be a torch.Tensor, got " f"{type(control_points).__name__}" ) if not isinstance(control_displacements, torch.Tensor): raise TypeError( "control_displacements must be a torch.Tensor, got " f"{type(control_displacements).__name__}" ) if point_weights is not None and not isinstance(point_weights, (str, tuple)): raise TypeError( "DomainMesh.radial_basis_function_deform point_weights must be " "a common point_data key/path, not a raw tensor" ) from physicsnemo.mesh.transformations.deform._utils import ( _resolve_domain_point_weights, ) from physicsnemo.nn.functional.geometry.deform import ( radial_basis_function_deform_points, ) components: list[tuple[str, Mesh]] = [("interior", self.interior)] components.extend( (f"boundaries[{name!r}]", self.boundaries[name]) for name in self.boundaries.keys() ) resolved_point_weights = _resolve_domain_point_weights( components, point_weights, control_points, "control_points" ) def apply_field( combined_points: Float[torch.Tensor, "n_points n_spatial_dims"], combined_point_weights: Bool[torch.Tensor, " n_points"] | Float[torch.Tensor, " n_points"] | None, ) -> Float[torch.Tensor, "n_points n_spatial_dims"]: return radial_basis_function_deform_points( combined_points, control_points, control_displacements, kernel=kernel, polynomial=polynomial, smoothing=smoothing, point_weights=combined_point_weights, implementation=implementation, ) return self._deform_components(components, resolved_point_weights, apply_field)
[docs] def free_form_deform( self, control_displacements: Float[ torch.Tensor, "*lattice_resolution n_spatial_dims" ], *, origin: Float[torch.Tensor, " n_spatial_dims"] | Sequence[builtins.float] | None = None, extent: Float[torch.Tensor, " n_spatial_dims"] | Sequence[builtins.float] | None = None, basis: _FFDBasis = "bernstein", point_weights: str | tuple[str, ...] | None = None, implementation: Literal["torch", "warp"] | None = None, ) -> "DomainMesh": """Deform the interior and all boundaries with one lattice field. Every component uses the same control lattice, box, basis, and backend. With ``point_weights=None``, coincident interior and boundary points receive the same motion. When supplied, ``point_weights`` is a common :attr:`Mesh.point_data` key (or nested tuple key) resolved independently on each component. Raw point-weight tensors are rejected because component point counts differ. Parameters ---------- control_displacements : torch.Tensor Displacement vectors, not destination coordinates, for every lattice node, with shape ``(n_1, ..., n_D, n_spatial_dims)`` and the same float32 or float64 dtype and device as every component's points. Each axis needs at least two nodes for ``"bernstein"`` and the node-interpolating bases, and four for ``"bspline"``. origin : torch.Tensor, sequence of float, or None, optional Minimum corner of the lattice box with shape ``(n_spatial_dims,)``. ``None`` uses the minimum corner of the combined component bounds. For repeated GPU calls with an explicit box, create ``origin`` and ``extent`` once as device tensors. Reuse them to avoid recreating and transferring sequence values. extent : torch.Tensor, sequence of float, or None, optional Edge lengths of the lattice box. Every value must be finite and strictly positive. The operation does not validate tensor values at runtime. ``None`` sizes the box from ``origin`` to the maximum corner of the combined component bounds. Validating a derived extent synchronizes with the device and is not CUDA Graph capture-safe. For capture, pass both ``origin`` and ``extent`` as device tensors. Every coordinate axis must have positive range when the extent is derived. Otherwise, supply an explicit extent. basis : {"bernstein", "bspline", "linear", "cubic_hermite", "quintic_hermite"}, optional Per-axis basis family. ``"bernstein"`` provides global support. ``"bspline"`` uses local four-node-per-axis support. B-spline coefficient index ``i`` corresponds to local coordinate ``(i - 1) / (n - 3)``. The first and last coefficient planes lie outside the evaluation box. ``"linear"``, ``"cubic_hermite"``, and ``"quintic_hermite"`` use two neighboring nodes per axis. The resulting fields are C0, C1, and C2 across cell boundaries, respectively. See :func:`~physicsnemo.mesh.transformations.deform.free_form_deform` for their polynomial weights and literature reference. Default is ``"bernstein"``. point_weights : str, tuple[str, ...], or None Optional point-data key present in every component and resolved independently on each component. Each resolved tensor must have shape ``(component.n_points,)`` and match the component point device. All components must use one common bool or floating dtype. Floating weights must also match the point dtype. Raw tensors are not accepted. implementation : {"torch", "warp"} or None Backend override. Automatic dispatch uses Torch on CPU. On CUDA, it uses Warp when available and otherwise Torch. Returns ------- DomainMesh New domain with deformed component meshes and unchanged domain data. Raises ------ TypeError If tensors, lattice values, or point weights have unsupported types or dtypes. ValueError If component layouts, lattice parameters, point weights, or ``basis`` are invalid. KeyError If a point-data key or ``implementation`` name is not found. ImportError If an explicitly requested backend is unavailable. Notes ----- The operation retains connectivity and attached mesh and domain data. It treats attached vector and tensor fields as Lagrangian data and does not push them forward. It invalidates geometry caches and retains topology caches on each component. Points outside the lattice box are unchanged. To keep the exterior fixed, zero the outermost coefficient plane on every Bernstein or node-interpolating face. For cubic B-splines, zero the first and last three coefficient planes on every axis. ``origin`` and ``extent`` are non-differentiable lattice parameters. The operation does not automatically detect inverted, degenerate, or self-intersecting cells. Validate each component mesh explicitly with :meth:`Mesh.validate` when required. """ if not isinstance(control_displacements, torch.Tensor): raise TypeError( "control_displacements must be a torch.Tensor, got " f"{type(control_displacements).__name__}" ) if point_weights is not None and not isinstance(point_weights, (str, tuple)): raise TypeError( "DomainMesh.free_form_deform point_weights must be a common " "point_data key/path, not a raw tensor" ) from physicsnemo.mesh.transformations.deform._utils import ( _resolve_domain_point_weights, ) from physicsnemo.mesh.transformations.deform.ffd import _default_lattice_box from physicsnemo.nn.functional.geometry.deform import free_form_deform_points components: list[tuple[str, Mesh]] = [("interior", self.interior)] components.extend( (f"boundaries[{name!r}]", self.boundaries[name]) for name in self.boundaries.keys() ) resolved_point_weights = _resolve_domain_point_weights( components, point_weights, control_displacements, "control_displacements" ) def apply_field( combined_points: Float[torch.Tensor, "n_points n_spatial_dims"], combined_point_weights: Bool[torch.Tensor, " n_points"] | Float[torch.Tensor, " n_points"] | None, ) -> Float[torch.Tensor, "n_points n_spatial_dims"]: box_origin, box_extent = _default_lattice_box( combined_points, origin, extent ) return free_form_deform_points( combined_points, control_displacements, origin=box_origin, extent=box_extent, basis=basis, point_weights=combined_point_weights, implementation=implementation, ) return self._deform_components(components, resolved_point_weights, apply_field)
def _deform_components( self, components: list[tuple[str, Mesh]], resolved_point_weights: list[ Bool[torch.Tensor, " n_component_points"] | Float[torch.Tensor, " n_component_points"] ], apply_field: Callable[ [ Float[torch.Tensor, "n_points n_spatial_dims"], Bool[torch.Tensor, " n_points"] | Float[torch.Tensor, " n_points"] | None, ], Float[torch.Tensor, "n_points n_spatial_dims"], ], ) -> "DomainMesh": """Deform every component with one combined world-space evaluation. Evaluating the common field once avoids repeated dispatch and field setup across boundaries. Splitting the result retains autograd links to every component's original points and optional point weights. """ component_meshes = [component for _, component in components] point_counts = [component.n_points for component in component_meshes] has_point_weights = bool(resolved_point_weights) if len(component_meshes) == 1: combined_points = component_meshes[0].points combined_point_weights = ( resolved_point_weights[0] if has_point_weights else None ) else: combined_points = torch.cat( [component.points for component in component_meshes], dim=0 ) combined_point_weights = ( torch.cat(resolved_point_weights, dim=0) if has_point_weights else None ) from physicsnemo.mesh.transformations.deform._utils import ( _mesh_with_deformed_points, ) combined_output = apply_field(combined_points, combined_point_weights) output_points = ( (combined_output,) if len(component_meshes) == 1 else combined_output.split(point_counts, dim=0) ) output_meshes = [ _mesh_with_deformed_points(component, points) for component, points in zip(component_meshes, output_points) ] interior = output_meshes[0] boundaries = { name: output_meshes[index] for index, name in enumerate(self.boundaries.keys(), start=1) } return DomainMesh( interior=interior, boundaries=boundaries, global_data=self.global_data.clone(), ) ### Cleanup / Refinement
[docs] def clean( self, tolerance: float = 1e-12, merge_points: bool = True, remove_duplicate_cells: bool = True, remove_unused_points: bool = True, ) -> "DomainMesh": r"""Clean and repair all meshes in the domain. Delegates to :meth:`Mesh.clean` for each mesh independently. Parameters ---------- tolerance : float, optional L2 distance threshold for merging duplicate points. merge_points : bool, optional Whether to merge spatially-duplicate points. remove_duplicate_cells : bool, optional Whether to remove cells with identical vertex sets. remove_unused_points : bool, optional Whether to drop points not referenced by any cell. Returns ------- DomainMesh New domain with cleaned meshes. """ return self.apply_to_meshes( lambda m: m.clean( tolerance=tolerance, merge_points=merge_points, remove_duplicate_cells=remove_duplicate_cells, remove_unused_points=remove_unused_points, ) )
[docs] def strip_caches(self) -> "DomainMesh": r"""Remove cached geometry from all meshes in the domain. Delegates to :meth:`Mesh.strip_caches` for each mesh. Returns ------- DomainMesh New domain with all cached values cleared. """ return self.apply_to_meshes(lambda m: m.strip_caches())
[docs] def subdivide( self, levels: int = 1, filter: Literal["linear", "butterfly", "loop"] = "linear", ) -> "DomainMesh": r"""Subdivide all meshes in the domain. Delegates to :meth:`Mesh.subdivide` for each mesh. Parameters ---------- levels : int, optional Number of subdivision iterations. filter : {"linear", "butterfly", "loop"}, optional Subdivision scheme. See :meth:`Mesh.subdivide`. Returns ------- DomainMesh New domain with subdivided meshes. """ return self.apply_to_meshes(lambda m: m.subdivide(levels=levels, filter=filter))
### Data Operations
[docs] def cell_data_to_point_data(self, overwrite_keys: bool = False) -> "DomainMesh": r"""Convert cell data to point data on all meshes in the domain. Delegates to :meth:`Mesh.cell_data_to_point_data` for each mesh. Parameters ---------- overwrite_keys : bool If ``True``, silently overwrite existing ``point_data`` keys. Returns ------- DomainMesh New domain with converted data on all meshes. """ return self.apply_to_meshes( lambda m: m.cell_data_to_point_data(overwrite_keys=overwrite_keys) )
[docs] def point_data_to_cell_data(self, overwrite_keys: bool = False) -> "DomainMesh": r"""Convert point data to cell data on all meshes in the domain. Delegates to :meth:`Mesh.point_data_to_cell_data` for each mesh. Parameters ---------- overwrite_keys : bool If ``True``, silently overwrite existing ``cell_data`` keys. Returns ------- DomainMesh New domain with converted data on all meshes. """ return self.apply_to_meshes( lambda m: m.point_data_to_cell_data(overwrite_keys=overwrite_keys) )
[docs] def compute_point_derivatives( self, keys: str | tuple[str, ...] | list[str | tuple[str, ...]] | None = None, method: Literal["lsq", "dec"] = "lsq", gradient_type: Literal["intrinsic", "extrinsic", "both"] = "intrinsic", ) -> "DomainMesh": r"""Compute gradients of point_data fields on all meshes. Delegates to :meth:`Mesh.compute_point_derivatives` for each mesh. Parameters ---------- keys : str or tuple or list or None, optional Fields to differentiate. ``None`` for all non-cached fields. method : {"lsq", "dec"}, optional Discretization method. gradient_type : {"intrinsic", "extrinsic", "both"}, optional Type of gradient to compute. Returns ------- DomainMesh Domain with gradient fields added to each mesh's ``point_data``. """ return self.apply_to_meshes( lambda m: m.compute_point_derivatives( keys=keys, method=method, gradient_type=gradient_type ) )
[docs] def compute_cell_derivatives( self, keys: str | tuple[str, ...] | list[str | tuple[str, ...]] | None = None, method: Literal["lsq", "dec"] = "lsq", gradient_type: Literal["intrinsic", "extrinsic", "both"] = "intrinsic", ) -> "DomainMesh": r"""Compute gradients of cell_data fields on all meshes. Delegates to :meth:`Mesh.compute_cell_derivatives` for each mesh. Parameters ---------- keys : str or tuple or list or None, optional Fields to differentiate. ``None`` for all non-cached fields. method : {"lsq", "dec"}, optional Discretization method. gradient_type : {"intrinsic", "extrinsic", "both"}, optional Type of gradient to compute. Returns ------- DomainMesh Domain with gradient fields added to each mesh's ``cell_data``. """ return self.apply_to_meshes( lambda m: m.compute_cell_derivatives( keys=keys, method=method, gradient_type=gradient_type ) )
### Validation
[docs] def validate( self, check_degenerate_cells: bool = True, check_duplicate_vertices: bool = True, check_inverted_cells: bool = False, check_out_of_bounds: bool = True, check_manifoldness: bool = False, tolerance: float | None = None, raise_on_error: bool = False, *, check_self_intersection: bool = False, ) -> dict[str, Any]: r"""Validate all meshes in the domain and aggregate results. Delegates to :meth:`Mesh.validate` for the interior and each boundary mesh, then aggregates the results into a domain-level report. Parameters ---------- check_degenerate_cells : bool, optional Check for zero/negative area cells. check_duplicate_vertices : bool, optional Check for coincident vertices. check_inverted_cells : bool, optional Check for negative orientation. check_out_of_bounds : bool, optional Check cell indices are valid. check_manifoldness : bool, optional Check manifold topology. tolerance : float | None, optional Tolerance for geometric checks. If ``None`` (default), each mesh uses a dtype-aware epsilon. raise_on_error : bool, optional Raise ``ValueError`` on first error vs return report. check_self_intersection : bool, optional Request self-intersection checks for every component. This option is keyword-only and not yet implemented; passing ``True`` raises ``NotImplementedError``. Returns ------- dict[str, Any] Aggregated validation report with keys: - ``"interior"``: validation report for the interior mesh (``Mapping[str, bool | int | torch.Tensor]``, see :meth:`Mesh.validate`). - ``"boundaries"``: ``dict[str, Mapping[str, ...]]`` of per-boundary reports. - ``"valid"``: ``bool``, ``True`` only if all meshes pass validation. Raises ------ NotImplementedError If ``check_self_intersection=True`` because component-level self-intersection checking is not yet implemented. """ kwargs: dict[str, Any] = dict( check_degenerate_cells=check_degenerate_cells, check_duplicate_vertices=check_duplicate_vertices, check_inverted_cells=check_inverted_cells, check_out_of_bounds=check_out_of_bounds, check_manifoldness=check_manifoldness, check_self_intersection=check_self_intersection, tolerance=tolerance, raise_on_error=raise_on_error, ) interior_report = self.interior.validate(**kwargs) boundary_reports = { name: self.boundaries[name].validate(**kwargs) for name in self.boundary_names } return { "interior": interior_report, "boundaries": boundary_reports, "valid": interior_report["valid"] and all(r["valid"] for r in boundary_reports.values()), }
### Properties @property def boundary_names(self) -> list[str]: """Sorted list of boundary condition names. Returns ------- list[str] The keys of ``boundaries``, sorted alphabetically. """ return sorted(self.boundaries.keys()) @property def n_boundaries(self) -> int: """Number of boundary meshes. Returns ------- int The number of entries in ``boundaries``. """ return len(self.boundary_names) ### Methods
[docs] def all_meshes(self) -> Iterator[tuple[str, Mesh]]: """Iterate over all meshes in the domain. Yields the interior mesh first (keyed ``"interior"``), then each boundary mesh in sorted key order. Yields ------ tuple[str, Mesh] ``(name, mesh)`` pairs. The first pair is always ``("interior", self.interior)``. Examples -------- >>> for name, mesh in dm.all_meshes(): ... print(f"{name}: {mesh.n_points} points") # doctest: +SKIP interior: 100 points inlet: 3 points no_slip: 3 points """ yield "interior", self.interior for name in self.boundary_names: yield name, self.boundaries[name]
def __iter__(self) -> Iterator[tuple[str, Mesh]]: r"""Iterate over all meshes in the domain. Equivalent to :meth:`all_meshes`; yields the interior mesh first (keyed ``"interior"``), then each boundary mesh in sorted key order. Yields ------ tuple[str, Mesh] ``(name, mesh)`` pairs. Examples -------- >>> for name, mesh in dm: ... print(f"{name}: {mesh.n_points} points") # doctest: +SKIP """ yield from self.all_meshes()
[docs] def merge_boundaries(self, preserve_data: bool = False) -> Mesh: """Merge all boundary meshes into a single :class:`Mesh`. Produces a mesh containing the concatenated points and cells from every boundary. By default, ``point_data`` and ``cell_data`` are stripped before merging because boundaries typically carry heterogeneous fields (different keys per boundary), which :meth:`Mesh.merge` cannot concatenate. Parameters ---------- preserve_data : bool If ``False`` (default), strip ``point_data`` and ``cell_data`` from each boundary before merging - the safe choice for the typical CFD case where each boundary carries its own field set. If ``True``, delegate directly to :meth:`Mesh.merge`, which preserves data but requires that all boundaries share the same ``cell_data`` keys and have ``point_data`` that can be concatenated. Use this when every boundary has a consistent set of fields. Returns ------- Mesh A single mesh containing the concatenated points and cells from every boundary. Data fields are included only if ``preserve_data`` is ``True``. Raises ------ ValueError If there are no boundary meshes to merge, if boundary meshes have incompatible manifold dimensions, or (when ``preserve_data=True``) if their data keys are inconsistent. """ if self.n_boundaries == 0: raise ValueError("No boundary meshes to merge.") boundaries = [self.boundaries[name] for name in self.boundary_names] if preserve_data: return Mesh.merge(boundaries) geometry_only = [Mesh(points=b.points, cells=b.cells) for b in boundaries] return Mesh.merge(geometry_only)
[docs] def is_boundary_watertight(self, tolerance: float = 1e-6) -> bool: r"""Check whether the merged boundary meshes form a watertight surface. Merges all boundary meshes via :meth:`merge_boundaries`, deduplicates coincident vertices with :meth:`Mesh.clean`, and calls :meth:`Mesh.is_watertight` on the result. The clean step is necessary because independently-meshed boundary patches share physical vertices that become duplicated during merge - and float32 round-off from any prior transform may prevent an exact-match merge. Parameters ---------- tolerance : float, optional L2 distance threshold for merging coincident boundary vertices before the topology check. The default ``1e-6`` is deliberately looser than :meth:`Mesh.clean`'s ``1e-12`` so it absorbs float32 round-off (~1e-7 relative) on the duplicated vertices that ``merge_boundaries`` produces from independently-meshed patches. For coordinates that span much smaller or much larger than ~1, pass an explicit value (e.g. ``1e-6 * max_extent`` of the bbox). Returns ------- bool ``True`` if the merged boundary surface is watertight (every codimension-1 facet is shared by exactly 2 cells), ``False`` otherwise. Returns ``False`` if there are no boundary meshes. Notes ----- This is not free to compute: the :meth:`Mesh.clean` step performs a BVH-based duplicate-point merge that scales as :math:`O(N \log N)` in the total boundary vertex count :math:`N`, and dominates the runtime. Callers that need the result repeatedly should cache it. """ if self.n_boundaries == 0: return False return self.merge_boundaries().clean(tolerance=tolerance).is_watertight()
[docs] def draw( self, *, backend: Literal["matplotlib", "pyvista", "auto"] = "auto", show: bool = True, point_scalars: None | torch.Tensor | str | tuple[str, ...] = None, cell_scalars: None | torch.Tensor | str | tuple[str, ...] = None, cmap: str = "viridis", vmin: float | None = None, vmax: float | None = None, alpha_points: float = 1.0, alpha_cells: float = 1.0, alpha_edges: float = 1.0, show_edges: bool = False, boundary_kwargs: dict[str, Any] | None = None, ax: "matplotlib.axes.Axes | pyvista.Plotter | None" = None, backend_options: dict[str, Any] | None = None, ) -> "matplotlib.axes.Axes | pyvista.Plotter": r"""Draw the domain: interior with optional scalar coloring, boundaries overlaid. Renders the interior as the primary visual layer, then overlays every boundary on the same canvas. The interior parameter set mirrors :meth:`Mesh.draw` exactly, with two intentional changes: the call is keyword-only, and ``show_edges`` defaults to ``False`` (rather than ``True``) because dense interior meshes are typically more readable without edges. Both matplotlib and PyVista backends are supported. Parameters ---------- backend, show, point_scalars, cell_scalars, cmap, vmin, vmax, alpha_points, alpha_cells, alpha_edges, show_edges, ax, backend_options Forwarded to :meth:`Mesh.draw` for the **interior** mesh. See :meth:`Mesh.draw` for full descriptions. boundary_kwargs : dict, optional Keyword arguments forwarded to :meth:`Mesh.draw` for **every** boundary mesh. Defaults are tuned for unobtrusive overlay: - ``alpha_points = 0`` (boundary vertices are not scattered). - ``alpha_cells = 0.3`` when boundaries are 2-D surfaces, ``1.0`` when they are 1-D curves. Auto-detected from the first boundary's :attr:`Mesh.n_manifold_dims`. - ``show_edges = False``. User-supplied keys override these defaults. To color individual boundaries by their own scalar fields, compose :meth:`Mesh.draw` calls directly (see Examples). Returns ------- matplotlib.axes.Axes or pyvista.Plotter The canvas, for further customization when ``show=False``. Examples -------- Default visualization with pressure coloring on the interior: >>> dm.draw(point_scalars="p", cmap="RdBu_r", vmin=-200, vmax=200) # doctest: +SKIP Translucent boundaries with edges visible: >>> dm.draw( # doctest: +SKIP ... point_scalars="p", ... boundary_kwargs={"alpha_cells": 0.5, "show_edges": True}, ... ) Customize and display later by setting axis limits on the returned canvas: >>> ax = dm.draw(point_scalars="p", show=False) # doctest: +SKIP >>> ax.set_xlim(-2, 4); ax.set_ylim(-3, 3) # doctest: +SKIP Per-boundary scalar coloring (manual composition - color the no-slip wall by its own ``shear`` field while the interior shows pressure): >>> ax = dm.interior.draw(point_scalars="p", show=False) # doctest: +SKIP >>> dm.boundaries["wall"].draw( # doctest: +SKIP ... ax=ax, cell_scalars="shear", cmap="hot", show=False, ... ) >>> for name in dm.boundary_names: # doctest: +SKIP ... if name == "wall": ... continue ... dm.boundaries[name].draw( ... ax=ax, alpha_cells=0.3, alpha_points=0, ... show_edges=False, show=False, ... ) >>> import matplotlib.pyplot as plt; plt.show() # doctest: +SKIP """ ### Auto-pick boundary opacity from the boundary's manifold dim: ### 2-D surfaces would otherwise occlude the interior; 1-D curves ### are thin lines and stay legible at full opacity. if self.n_boundaries > 0: first_bdy = self.boundaries[self.boundary_names[0]] auto_alpha_cells = 0.3 if first_bdy.n_manifold_dims >= 2 else 1.0 else: auto_alpha_cells = 1.0 # unused boundary_defaults: dict[str, Any] = { "alpha_points": 0, "alpha_cells": auto_alpha_cells, "show_edges": False, } boundary_defaults.update(boundary_kwargs or {}) ### Draw interior; if no boundaries follow, this is the layer that ### triggers the eventual ``.show()``. has_boundaries = self.n_boundaries > 0 canvas = self.interior.draw( backend=backend, show=show and not has_boundaries, point_scalars=point_scalars, cell_scalars=cell_scalars, cmap=cmap, vmin=vmin, vmax=vmax, alpha_points=alpha_points, alpha_cells=alpha_cells, alpha_edges=alpha_edges, show_edges=show_edges, ax=ax, backend_options=backend_options, ) ### Overlay boundaries; the last one triggers ``.show()`` if requested. names = self.boundary_names last = names[-1] if names else None for name in names: self.boundaries[name].draw( ax=canvas, backend=backend, show=show and (name is last), **boundary_defaults, ) return canvas
### Repr is defined after the class body (see below) because ### @tensorclass overwrites __repr__ even when defined inline. ### Override the tensorclass __repr__ with custom formatting. # Must be done after class definition because @tensorclass overrides __repr__ # even when defined inside the class body (same pattern as Mesh). def _domain_mesh_repr(self: DomainMesh) -> str: """Format a readable summary of the domain mesh.""" lines = ["DomainMesh("] ### Interior - indent data fields one level under "interior:" interior_repr = format_mesh_repr(self.interior) first, *rest = interior_repr.split("\n") lines.append(f" interior: {first}") lines.extend(f" {line}" for line in rest) ### Boundaries - indent data fields one level under each boundary key bc_names = self.boundary_names if not bc_names: lines.append(" boundaries: {}") else: lines.append(" boundaries:") max_bc_len = max(len(n) for n in bc_names) for name in bc_names: bc_mesh = self.boundaries[name] bc_repr = format_mesh_repr(bc_mesh) first, *rest = bc_repr.split("\n") lines.append(f" {name.ljust(max_bc_len)}: {first}") lines.extend(f" {line}" for line in rest) ### Global data (only if non-empty) gd_keys = sorted(self.global_data.keys()) if gd_keys: items = ", ".join(f"{k}: {tuple(self.global_data[k].shape)}" for k in gd_keys) lines.append(f" global_data: {{{items}}}") lines.append(")") return "\n".join(lines) DomainMesh.__repr__ = _domain_mesh_repr # type: ignore[method-assign] # ty: ignore[invalid-assignment] ### Override the tensorclass ``to`` for the same reason as ``Mesh.to``: a floating/ # complex dtype cast via the generated tensorclass ``to`` recurses into the interior/ # boundary meshes and casts their integer ``cells`` to a float dtype, which fails # ``Mesh.__post_init__``. Only an explicitly requested floating dtype takes the # per-mesh path through the (cells-safe) ``Mesh.to`` via ``apply_to_meshes`` (with # ``global_data`` cast too); device-only moves and non-float dtypes are delegated # unchanged (cells-safe and metadata-preserving). def _domain_mesh_to(self, *args: Any, **kwargs: Any) -> "DomainMesh": cast_dtype = _requested_float_dtype(args, kwargs) if cast_dtype is None: return _tensorclass_domain_to(self, *args, **kwargs) # Per-mesh: route through the (fixed, cells-safe) ``Mesh.to``. Resolve the target # device with a zero-length probe, then move ``global_data`` to that device # (forwarding all transfer options except ``dtype``) and cast its floating leaves. probe = self.interior.points[:0].to(*args, **kwargs) moved = self.apply_to_meshes(lambda mesh: mesh.to(*args, **kwargs)) transfer_kwargs = {k: v for k, v in kwargs.items() if k != "dtype"} transfer_kwargs["device"] = probe.device moved.global_data = moved.global_data.to(**transfer_kwargs).apply( lambda t: t.to(cast_dtype) if (t.is_floating_point() or t.is_complex()) else t ) return moved _tensorclass_domain_to = DomainMesh.to # the generated tensorclass ``to`` DomainMesh.to = _domain_mesh_to # type: ignore[method-assign] # ty: ignore[invalid-assignment]