fenicsx_cosim package

Subpackages

fenicsx_cosim.adapters package

Submodules

fenicsx_cosim.communicator module

Communicator — ZeroMQ-based inter-process communication engine.

This module wraps pyzmq to handle the actual network communication between coupled FEniCSx solver processes. It is analogous to the communication backends in Kratos CoSimIO (file-based, socket-based) but uses ZeroMQ exclusively, which:

Does not interfere with FEniCSx’s internal MPI communicators.
Supports TCP/IP for distributed runs and IPC (Unix sockets) for same-machine runs.
Handles serialization of NumPy arrays via the helpers in utils.

Typical usage

>>> comm = Communicator(name="ThermalSolver", partner_name="MechSolver",
...                     role="connect", endpoint="tcp://localhost:5555")
>>> comm.send_array("Temperature", temp_array)
>>> name, disp_array = comm.receive_array()
>>> comm.close()

exception fenicsx_cosim.communicator.CommunicationError[source]

Bases: Exception

Raised when a communication operation fails.

class fenicsx_cosim.communicator.Communicator(name: str, partner_name: str, role: Literal['bind', 'connect'], endpoint: str = 'tcp://localhost:5555', timeout_ms: int = 30000, handshake: bool = True)[source]

Bases: object

ZeroMQ-based communicator for inter-process data exchange.

Two coupled solvers each create a Communicator. One must take the "bind" role (the server) and the other the "connect" role (the client). The underlying socket type is zmq.PAIR which gives bidirectional, exclusive communication — ideal for a two-solver coupling scenario.

Parameters:

name (str) – Identifier for this solver.
partner_name (str) – Expected identifier of the partner solver.
role ({"bind", "connect"}) – Whether this communicator binds (acts as server) or connects (acts as client).
endpoint (str, optional) – A ZeroMQ endpoint string. Defaults to "tcp://localhost:5555". For same-machine IPC use "ipc:///tmp/fenicsx_cosim".
timeout_ms (int, optional) – Timeout in milliseconds for blocking receives. Default 30 000.
handshake (bool, optional) – If True (default), perform a name-exchange handshake upon initialization to verify that both sides are correctly paired.

close() → None[source]: Gracefully close the ZeroMQ socket.

property is_connected: bool: Whether the communicator socket is still open.

receive_array() → tuple[str, np.ndarray | 'sps.csr_matrix'][source]

Receive a named NumPy array from the partner.

Returns:: (data_name, array)
Return type:: tuple[str, np.ndarray | sps.csr_matrix]
Raises:: CommunicationError – If a timeout occurs or the socket is closed.

receive_raw() → bytes[source]: Receive raw bytes from the partner.

send_array(data_name: str, array: np.ndarray | 'sps.csr_matrix') → None[source]

Serialize and send a named NumPy array to the partner.

Parameters:

data_name (str) – Identifier for the data (e.g. "TemperatureField").
array (np.ndarray or scipy.sparse.spmatrix) – The data to transmit.

Raises:

CommunicationError – If the socket is closed or the send fails.

send_raw(data: bytes) → None[source]: Send raw bytes to the partner.

synchronize() → None[source]

Block until both solvers reach this synchronization point.

The bind side sends the sync signal first, then waits for ACK. The connect side waits for the sync signal, then sends ACK.

fenicsx_cosim.coupling_interface module

CouplingInterface — The main user-facing API for fenicsx-cosim.

This class integrates the MeshExtractor, Communicator, and DataMapper into a single, clean entry point that hides all networking and mapping details from the user.

It is the FEniCSx analogue of the top-level CoSimIO.Connect / ExportData / ImportData workflow in Kratos CoSimIO.

Additionally, it now supports:

Adaptive Mesh Refinement (AMR): via update_interface_geometry() which re-negotiates the mapping with the partner solver.
FE² Multiscale Homogenization: via register_quadrature_space(), scatter_data(), and gather_data() for dispatching integration- point tensors to a pool of RVE solver workers.

Typical usage (Standard Boundary Coupling)

>>> from fenicsx_cosim import CouplingInterface
>>> cosim = CouplingInterface(
...     name="ThermalSolver",
...     partner_name="MechanicalSolver",
... )
>>> cosim.register_interface(mesh, facet_tags, marker_id=1, function_space=V)
>>> cosim.export_data("TemperatureField", temperature)
>>> cosim.import_data("DisplacementField", displacement)
>>> cosim.advance_in_time()

Typical usage (FE² Scatter-Gather)

>>> cosim_fe2 = CouplingInterface(name="Macro", role="Master",
...     topology="scatter-gather")
>>> cosim_fe2.register_quadrature_space(V_quad)
>>> cosim_fe2.scatter_data("StrainTensor", macro_strains)
>>> stresses = cosim_fe2.gather_data("StressTensor")

class fenicsx_cosim.coupling_interface.CouplingInterface(name: str, partner_name: str = 'Workers', role: str | None = None, endpoint: str | None = None, connection_type: str = 'tcp', timeout_ms: int = 30000, enable_mapping: bool = True, topology: str = 'pair', push_endpoint: str = 'tcp://*:5556', pull_endpoint: str = 'tcp://*:5557')[source]

Bases: object

High-level co-simulation coupling API for FEniCSx solvers.

This is the single entry point that researchers interact with. It orchestrates boundary-data extraction, inter-process communication, and (optionally) non-conforming mesh mapping.

Parameters:

name (str) – A unique identifier for this solver instance (e.g. "ThermalSolver").
partner_name (str) – The identifier of the partner solver. For scatter-gather topology, this can be set to "Workers".
role ({"bind", "connect", "Master", "Worker"}, optional) – ZeroMQ role. The solver that starts first should "bind"; the one that starts second should "connect". If None (default), the role is determined automatically by sorting the names lexicographically — the name that comes first binds. Use "Master" / "Worker" for scatter-gather topology.
endpoint (str, optional) – ZeroMQ endpoint. Default "tcp://localhost:5555".
connection_type ({"tcp", "ipc"}, optional) – Shortcut to auto-generate an endpoint. Ignored if endpoint is explicitly provided.
timeout_ms (int, optional) – Receive timeout in ms (default 30 000).
enable_mapping (bool, optional) – If True (default), a NearestNeighborMapper is built automatically when both sides exchange boundary coordinates.
topology ({"pair", "scatter-gather"}, optional) – Communication topology. "pair" (default) uses the standard zmq.PAIR socket for 1-to-1 coupling. "scatter-gather" uses PUSH/PULL for FE² fan-out / fan-in.
push_endpoint (str, optional) – For scatter-gather only: PUSH socket endpoint.
pull_endpoint (str, optional) – For scatter-gather only: PULL socket endpoint.

advance_adapter() → None[source]: Synchronize and advance the adapter’s solver.

advance_in_time() → None[source]

Synchronize both solvers at the end of a time step.

This acts as a barrier — neither solver proceeds until both have called advance_in_time().

property boundary_coordinates: ndarray: Boundary DoF coordinates of this solver — shape (N, 3).

check_mesh_update() → bool[source]

Lightweight check to see if the partner solver updated its mesh.

If the partner calls update_interface_geometry(), this solver must call check_mesh_update() at the same point in the time loop to receive the new coordinates and re-build the mapping.

If neither solver refined its mesh but AMR is active, both solvers should call check_mesh_update() to acknowledge no changes.

Returns:: True if the partner updated its mesh and the mapping was rebuilt; False otherwise.
Return type:: bool

disconnect() → None[source]: Gracefully close the connection to the partner solver.

property dynamic_mapper: DynamicMapper | None: The dynamic (AMR-aware) mapper, if active.

export_data(data_name: str, function: dolfinx.fem.Function) → None[source]

Extract boundary values from a FEniCSx Function and send them to the partner solver.

Parameters:

data_name (str) – Identifier for the data (e.g. "TemperatureField").
function (dolfinx.fem.Function) – The FEniCSx function whose boundary values are to be sent.

export_raw(data_name: str, array: np.ndarray | 'sps.csr_matrix') → None[source]

Send a raw NumPy array or SciPy sparse matrix to the partner solver.

This is useful for sending non-field data like sparse stiffness matrices or scalar values (e.g. for Shakedown analysis or custom optimization loops).

Parameters:

data_name (str) – Identifier for the data.
array (np.ndarray or scipy.sparse.spmatrix) – The raw data to transmit.

export_via_adapter(field_name: str) → None[source]

Extract field from adapter and send to partner.

Parameters:: field_name (str) – Field name as understood by the adapter.

property extractor: MeshExtractor: The underlying mesh extractor.

classmethod from_adapter(adapter: SolverAdapter, name: str, partner_name: str = 'Partner', role: str | None = None, endpoint: str | None = None, connection_type: str = 'tcp', timeout_ms: int = 30000, enable_mapping: bool = True) → CouplingInterface[source]

Create a CouplingInterface using any SolverAdapter.

This allows non-FEniCSx solvers (Kratos, Abaqus, etc.) to participate in fenicsx-cosim coupling via their respective adapters.

Parameters:

adapter (SolverAdapter) – A concrete adapter for the solver.
name (str) – Identifier for this solver.
partner_name (str, optional) – Identifier of the partner solver.
role ({"bind", "connect"}, optional) – ZeroMQ role. Auto-determined if None.
endpoint (str, optional) – ZeroMQ endpoint.
connection_type ({"tcp", "ipc"}, optional) – Transport type.
timeout_ms (int, optional) – Receive timeout.
enable_mapping (bool, optional) – Build nearest-neighbour mapper.

Return type:

CouplingInterface

gather_data(data_name: str, function: dolfinx.fem.Function | None = None, n_expected: int | None = None) → list[ndarray][source]

Gather results from the RVE worker pool and optionally inject them into a Quadrature function.

Parameters:

data_name (str) – Name identifier (e.g. "StressTensor").
function (dolfinx.fem.Function, optional) – If provided, the gathered values are injected into this Quadrature function.
n_expected (int, optional) – Number of results to wait for. Defaults to quadrature_data.num_cells.

Returns:

The gathered per-cell result arrays.

Return type:

list[np.ndarray]

import_data(data_name: str, function: dolfinx.fem.Function) → None[source]

Receive boundary values from the partner solver and inject them into a FEniCSx Function.

If the meshes are non-conforming (different DoF coordinates), the received values are first mapped using the DataMapper before injection.

Parameters:

data_name (str) – Expected identifier for the incoming data.
function (dolfinx.fem.Function) – The target FEniCSx function.

import_raw(data_name: str) → np.ndarray | 'sps.csr_matrix'[source]

Receive a raw NumPy array or SciPy sparse matrix from the partner solver.

Parameters:: data_name (str) – Expected identifier for the incoming data.
Returns:: The raw data received.
Return type:: np.ndarray or scipy.sparse.spmatrix

import_via_adapter(field_name: str) → None[source]

Receive field from partner and inject into adapter.

Parameters:: field_name (str) – Expected field name.

property mapper: DataMapper | None: The data mapper (if mapping is enabled and built).

property partner_coordinates: ndarray | None: Boundary DoF coordinates received from the partner.

property quadrature_extractor: QuadratureExtractor | None: The quadrature extractor, if registered.

register_adapter_interface() → None[source]

Register the coupling interface using the attached adapter.

Extracts boundary coordinates from the adapter and exchanges them with the partner solver to build the data mapping.

Raises:: RuntimeError – If no adapter has been set (use from_adapter()).

register_interface(mesh: dolfinx.mesh.Mesh, facet_tags: dolfinx.mesh.MeshTags, marker_id: int, function_space: dolfinx.fem.FunctionSpace | None = None) → None[source]

Register the coupling boundary and exchange coordinates with the partner solver.

Parameters:

mesh (dolfinx.mesh.Mesh) – The computational mesh.
facet_tags (dolfinx.mesh.MeshTags) – Mesh tags labelling boundary facets.
marker_id (int) – The integer marker for the coupling boundary.
function_space (dolfinx.fem.FunctionSpace, optional) – The function space. If None, a default scalar Lagrange-1 space is created on mesh.

register_interface_from_locator(mesh: dolfinx.mesh.Mesh, locator_fn, function_space: dolfinx.fem.FunctionSpace | None = None, marker_id: int = 0) → None[source]

Register the coupling boundary using a geometric locator function.

Parameters:

mesh (dolfinx.mesh.Mesh) – The computational mesh.
locator_fn (callable) – locator_fn(x: ndarray) -> ndarray[bool]
function_space (dolfinx.fem.FunctionSpace, optional) – Defaults to Lagrange-1.
marker_id (int, optional) – Label for this boundary.

register_quadrature_space(function_space_or_mesh=None, quadrature_degree: int = 2, tensor_shape=()) → None[source]

Register a Quadrature function space for FE² homogenization.

This replaces the standard boundary registration and instead sets up data exchange at integration points.

Parameters:

function_space_or_mesh (dolfinx.fem.FunctionSpace or dolfinx.mesh.Mesh) – Either a pre-existing Quadrature function space, or a mesh on which to create one.
quadrature_degree (int, optional) – If a mesh is provided, the quadrature degree. Default 2.
tensor_shape (tuple[int, ...], optional) – Shape of the tensor at each integration point.

scatter_data(data_name: str, function: dolfinx.fem.Function, metadata: list[dict] | None = None) → int[source]

Extract per-cell integration-point values and scatter them to the RVE worker pool.

Parameters:

data_name (str) – Name identifier (e.g. "StrainTensor").
function (dolfinx.fem.Function) – Source function on the Quadrature space.
metadata (list[dict], optional) – Optional per-cell metadata (material params, damage state).

Returns:

Number of work items dispatched.

Return type:

int

scatter_gather_data(scatter_name: str, scatter_function: dolfinx.fem.Function, gather_name: str, gather_function: dolfinx.fem.Function | None = None, metadata: list[dict] | None = None) → list[ndarray][source]

Convenience: scatter, then gather in one call.

Parameters:

scatter_name (str) – Name for the outgoing data.
scatter_function (dolfinx.fem.Function) – Source Quadrature function.
gather_name (str) – Name for the incoming data.
gather_function (dolfinx.fem.Function, optional) – Target Quadrature function for injection.
metadata (list[dict], optional) – Per-cell metadata.

Returns:

Gathered results.

Return type:

list[np.ndarray]

property step_count: int: Number of completed time steps.

update_interface_geometry(mesh: dolfinx.mesh.Mesh, facet_tags: dolfinx.mesh.MeshTags, marker_id: int, function_space: dolfinx.fem.FunctionSpace | None = None) → None[source]

Notify the coupling system that the mesh has been refined (AMR).

This method:

Re-extracts boundary DoFs from the refined mesh.
Invalidates the current data mapping.
Negotiates a synchronized coordinate exchange with the partner solver.
Rebuilds the DynamicMapper with the updated point clouds.

Parameters:

mesh (dolfinx.mesh.Mesh) – The newly refined mesh.
facet_tags (dolfinx.mesh.MeshTags) – Updated facet tags for the refined mesh.
marker_id (int) – The coupling boundary marker.
function_space (dolfinx.fem.FunctionSpace, optional) – Updated function space (if None, a new Lagrange-1 space is created on the refined mesh).

fenicsx_cosim.data_mapper module

DataMapper — Interpolation engine for non-conforming mesh boundaries.

When two coupled solvers use different meshes on the coupling boundary, the DoF coordinates will not coincide. The DataMapper bridges this gap by computing a mapping between the two point clouds and interpolating field values accordingly.

This corresponds to Phase 4 of the development roadmap and mirrors the mapping capabilities provided by the Kratos MappingApplication.

Currently implemented strategies:

NearestNeighborMapper — for each target point, find the closest source point and copy its value. Uses scipy.spatial.KDTree for O(N log N) performance.

Planned (future scope):

ProjectionMapper — Galerkin projection between non-matching boundary meshes.

Typical usage

>>> mapper = NearestNeighborMapper()
>>> mapper.build(source_coords, target_coords)
>>> mapped_values = mapper.map(source_values)

class fenicsx_cosim.data_mapper.DataMapper[source]

Bases: ABC

Abstract base class for all data mapping strategies.

Subclasses must implement build() and map().

abstractmethod build(source_coords: ndarray, target_coords: ndarray) → None[source]

Build the mapping from source to target point clouds.

Parameters:

source_coords (np.ndarray) – Coordinates of the source DoFs, shape (N_src, 3).
target_coords (np.ndarray) – Coordinates of the target DoFs, shape (N_tgt, 3).

abstractmethod inverse_map(target_values: ndarray) → ndarray[source]

Map values from target DoFs back to source DoFs.

Parameters:: target_values (np.ndarray) – Field values at the target DoFs.
Returns:: Interpolated values at the source DoFs.
Return type:: np.ndarray

abstractmethod map(source_values: ndarray) → ndarray[source]

Map values from source DoFs to target DoFs.

Parameters:: source_values (np.ndarray) – Field values at the source DoFs, shape (N_src,) or (N_src, D) for vector fields.
Returns:: Interpolated values at the target DoFs.
Return type:: np.ndarray

class fenicsx_cosim.data_mapper.NearestNeighborMapper[source]

Bases: DataMapper

Maps data by assigning each target DoF the value of its nearest source DoF.

This is the simplest (and often most robust) mapping strategy. It is suitable when the two meshes have similar resolution on the coupling boundary and the fields vary smoothly.

max_distance

After building, the maximum distance between any target point and its nearest source point. Useful for diagnosing mesh mismatch issues.

Type:: float or None

build(source_coords: ndarray, target_coords: ndarray) → None[source]

Build the nearest-neighbor mapping.

Parameters:

source_coords (np.ndarray) – Shape (N_src, 3).
target_coords (np.ndarray) – Shape (N_tgt, 3).

property forward_distances: ndarray: Distances between each target point and its nearest source point.

inverse_map(target_values: ndarray) → ndarray[source]

Map values from target DoFs back to source DoFs.

Parameters:: target_values (np.ndarray) – Shape (N_tgt,) or (N_tgt, D).
Returns:: Shape (N_src,) or (N_src, D).
Return type:: np.ndarray

map(source_values: ndarray) → ndarray[source]

Map values from source DoFs to target DoFs using nearest neighbor.

Parameters:: source_values (np.ndarray) – Shape (N_src,) for scalar or (N_src, D) for vector fields.
Returns:: Shape (N_tgt,) or (N_tgt, D).
Return type:: np.ndarray

fenicsx_cosim.dynamic_mapper module

DynamicMapper — Adaptive mesh-aware data mapping for AMR & Shakedown.

Standard co-simulation assumes a static coupling interface: the boundary coordinates are exchanged once during register_interface() and never change. When Adaptive Mesh Refinement (AMR) is used (dolfinx.mesh.refine), the DoF coordinates change, rendering the previous scipy.spatial.KDTree and its mapping indices invalid.

The DynamicMapper solves this by:

Detecting that the underlying mesh has changed (via a revision counter or explicit user call).
Broadcasting an UPDATE_MESH signal over ZeroMQ to the partner solver so both processes pause, exchange their newly-refined boundary point clouds, and rebuild the interpolation mapping.
Rebuilding the KDTree and mapping indices atomically, ensuring the next export_data / import_data call uses the updated geometry.

This corresponds to Section 1.1 of the Advanced Features Addendum in the project specification.

Typical usage (inside a time loop)

>>> if error_too_high:
...     domain = dolfinx.mesh.refine(domain, markers)
...     cosim.update_interface_geometry(domain, facet_tags, marker_id=1)
>>> cosim.export_data("Temperature", temperature)

class fenicsx_cosim.dynamic_mapper.DynamicMapper(base_mapper_cls: type[DataMapper] = <class 'fenicsx_cosim.data_mapper.NearestNeighborMapper'>)[source]

Bases: object

Mesh-change–aware wrapper around DataMapper.

The DynamicMapper keeps track of a revision counter which is bumped every time the mesh changes. It provides helpers to negotiate a synchronized re-mapping with the partner solver.

Parameters:: base_mapper_cls (type[DataMapper], optional) – The concrete mapper to use. Defaults to NearestNeighborMapper.

revision

The number of times the mapping has been (re-)built.

Type:: int

max_distance

Delegates to the underlying DataMapper.max_distance.

Type:: float or None

build(source_coords: ndarray, target_coords: ndarray) → None[source]

Build (or rebuild) the underlying mapping.

Parameters:

source_coords (np.ndarray) – Partner (source) boundary coordinates — shape (N_src, 3).
target_coords (np.ndarray) – Local (target) boundary coordinates — shape (N_tgt, 3).

invalidate() → None[source]

Mark the current mapping as stale.

This should be called whenever the local mesh has been refined or changed. The mapping will not be usable until build() is called again.

inverse_map(target_values: ndarray) → ndarray[source]: Map values from local (target) back to partner (source) DoFs.

property local_coordinates: ndarray | None: The local boundary coordinates used in the latest mapping.

map(source_values: ndarray) → ndarray[source]

Map values from partner (source) to local (target) DoFs.

Raises RuntimeError if the mapping is stale.

property max_distance: float | None: Maximum mapping distance from the underlying mapper.

property needs_update: bool: True if the mapping has been invalidated and must be rebuilt.

negotiate_update(communicator, role: str, my_new_coords: ndarray | None) → ndarray | None[source]

Negotiate a mesh-update with the partner solver.

Both solvers must call this method at the same point in the time loop. The protocol:

The bind side sends either UPDATE_MESH_SIGNAL or NO_UPDATE_SIGNAL.
The connect side receives the signal and then sends its own signal (likewise UPDATE_MESH_SIGNAL or NO_UPDATE_SIGNAL).
If either side signals an update, both exchange new boundary coordinates and return the partner’s new coordinates. Otherwise None is returned and the mapping is untouched.

Parameters:

communicator (Communicator) – The ZeroMQ communicator already connected to the partner.
role (str) – "bind" or "connect".
my_new_coords (np.ndarray or None) – The new local boundary coordinates if the mesh changed, or None if no change occurred.

Returns:

The partner’s new boundary coordinates if an exchange took place, or None if no update was needed.

Return type:

np.ndarray or None

property partner_coordinates: ndarray | None: The partner boundary coordinates used in the latest mapping.

fenicsx_cosim.mesh_extractor module

MeshExtractor — FEniCSx boundary data extraction engine.

This module handles the highly specific FEniCSx v0.10+ API calls required to isolate boundary Degrees of Freedom (DoFs) and their spatial coordinates from a dolfinx.mesh.Mesh.

It corresponds to Phase 1 of the development roadmap and is analogous to the CoSimIO::ModelPart concept in Kratos, but works natively with FEniCSx data structures.

Typical usage

>>> extractor = MeshExtractor()
>>> extractor.register(mesh, facet_tags, marker_id=1, function_space=V)
>>> coords = extractor.boundary_coordinates   # (N, 3) ndarray
>>> dof_indices = extractor.boundary_dof_indices  # (N,) ndarray

class fenicsx_cosim.mesh_extractor.BoundaryData(dof_indices: ndarray = <factory>, coordinates: ndarray = <factory>, facet_indices: ndarray = <factory>, marker_id: int = 0)[source]

Bases: object

Container for extracted boundary information.

dof_indices

Global DoF indices that live on the coupling boundary. Shape (N,).

Type:: np.ndarray

coordinates

Spatial coordinates of the boundary DoFs. Shape (N, 3).

Type:: np.ndarray

facet_indices

Mesh facet indices that form the coupling boundary.

Type:: np.ndarray

marker_id

The marker value used to identify the coupling boundary.

Type:: int

coordinates: ndarray

dof_indices: ndarray

facet_indices: ndarray

marker_id: int = 0

class fenicsx_cosim.mesh_extractor.MeshExtractor[source]

Bases: object

Extract boundary DoFs and coordinates from a FEniCSx mesh.

This class wraps the FEniCSx v0.10+ API to:

Identify boundary facets using mesh-tags or a locator function.
Find the DoFs living on those facets via topological DoF location.
Extract the (x, y, z) coordinates of the boundary DoFs.

The extracted data is stored internally and can be queried as NumPy arrays for serialization and transmission via the Communicator.

property boundary_coordinates: ndarray: Spatial coordinates of boundary DoFs — shape (N, 3).

property boundary_data: BoundaryData: The most recently extracted BoundaryData.

property boundary_dof_indices: ndarray: Global DoF indices on the coupling boundary — shape (N,).

extract_boundary_values(function: dolfinx.fem.Function) → ndarray[source]

Extract the values of a FEniCSx Function at the boundary DoFs.

Parameters:: function (dolfinx.fem.Function) – The FEniCSx function whose boundary values are needed.
Returns:: Values at the boundary DoFs.
Return type:: np.ndarray

inject_boundary_values(function: dolfinx.fem.Function, values: ndarray) → None[source]

Inject values into a FEniCSx Function at the boundary DoFs.

Parameters:

function (dolfinx.fem.Function) – The target FEniCSx function.
values (np.ndarray) – The values to inject (must match the number of boundary DoFs).

Raises:

ValueError – If values length does not match the number of boundary DoFs.

register(mesh: dolfinx.mesh.Mesh, facet_tags: dolfinx.mesh.MeshTags, marker_id: int, function_space: dolfinx.fem.FunctionSpace) → BoundaryData[source]

Register a coupling boundary and extract its DoF information.

Parameters:

mesh (dolfinx.mesh.Mesh) – The computational mesh.
facet_tags (dolfinx.mesh.MeshTags) – Mesh tags that label boundary facets.
marker_id (int) – The integer marker identifying the coupling boundary within facet_tags.
function_space (dolfinx.fem.FunctionSpace) – The function space whose DoFs on the boundary are needed.

Returns:

A dataclass holding the extracted boundary information.

Return type:

BoundaryData

register_from_locator(mesh: dolfinx.mesh.Mesh, locator_fn, function_space: dolfinx.fem.FunctionSpace, marker_id: int = 0) → BoundaryData[source]

Register a coupling boundary using a geometric locator function.

This is a convenience method for cases where no MeshTags object exists. The locator_fn must accept an (3, N) array of coordinates and return a boolean mask of length N.

Parameters:

mesh (dolfinx.mesh.Mesh) – The computational mesh.
locator_fn (callable) – locator_fn(x: ndarray) -> ndarray[bool] — identifies boundary facets.
function_space (dolfinx.fem.FunctionSpace) – The function space whose DoFs on the boundary are needed.
marker_id (int, optional) – A label for this boundary (default 0).

Return type:

BoundaryData

fenicsx_cosim.quadrature_extractor module

QuadratureExtractor — Integration-point data extraction for FE² homogenization.

In multiscale FE² homogenization, data is not exchanged at the mesh boundaries but at the integration (Gauss) points of every element. The macroscopic solver computes a strain tensor at each integration point and dispatches it to a Representative Volume Element (RVE) solver. The RVE solver returns the homogenized stress tensor and tangent stiffness.

The QuadratureExtractor provides the FEniCSx v0.10+ API calls to:

Create and manage Quadrature function spaces:

V_quad = dolfinx.fem.functionspace(mesh, ("Quadrature", degree, shape))

Extract tensor values (e.g. macroscopic strain) from Quadrature functions.
Inject tensor values (e.g. homogenized stress) back in, with correct element → global DoF mapping.

This corresponds to Section 1.2 of the Advanced Features Addendum.

Typical usage

>>> extractor = QuadratureExtractor()
>>> extractor.register(mesh, quadrature_degree=2, tensor_shape=(3, 3))
>>> strains = extractor.extract_values(macro_strain_function)
>>> extractor.inject_values(stress_function, homogenized_stresses)

class fenicsx_cosim.quadrature_extractor.QuadratureData(num_cells: int = 0, points_per_cell: int = 0, total_points: int = 0, tensor_shape: tuple = (), dof_per_value: int = 1, coordinates: ndarray = <factory>)[source]

Bases: object

Container for quadrature-point information.

num_cells

Number of mesh cells (elements).

Type:: int

points_per_cell

Number of quadrature/integration points per cell.

Type:: int

total_points

num_cells * points_per_cell.

Type:: int

tensor_shape

Shape of each tensor at each integration point (e.g. (3, 3) for a 2D symmetric stress/strain tensor, or (6,) for Voigt).

Type:: tuple[int, …]

dof_per_value

Number of scalar DoFs per integration point (product of tensor_shape).

Type:: int

coordinates

Physical coordinates of all integration points — shape (total_points, gdim).

Type:: np.ndarray

coordinates: ndarray

dof_per_value: int = 1

num_cells: int = 0

points_per_cell: int = 0

tensor_shape: tuple = ()

total_points: int = 0

class fenicsx_cosim.quadrature_extractor.QuadratureExtractor[source]

Bases: object

Extract and inject tensor data at integration points.

This class works with FEniCSx Quadrature function spaces to handle the data exchange required for FE² (computational homogenization).

The user creates a QuadratureExtractor, registers it with a mesh and quadrature degree, and then uses it to extract/inject tensors that are stored as dolfinx.fem.Function objects on the Quadrature space.

property cell_to_dof_map: ndarray: Cell → global DoF index mapping — shape (num_cells, dofs_per_cell).

extract_cell_values(function: dolfinx.fem.Function, cell_index: int) → ndarray[source]

Extract values for a single cell (element).

Parameters:

function (dolfinx.fem.Function) – A function on the Quadrature space.
cell_index (int) – The local cell index.

Returns:

Values at the integration points of the specified cell. Shape (points_per_cell,) for scalar, or (points_per_cell, *tensor_shape) for tensors.

Return type:

np.ndarray

extract_for_dispatch(function: dolfinx.fem.Function) → list[ndarray][source]

Extract per-cell integration-point values for dispatching to RVEs.

Returns a list of num_cells arrays, each of shape (points_per_cell,) or (points_per_cell, *tensor_shape). Cell i corresponds to RVE i.

Parameters:: function (dolfinx.fem.Function) – Source function on the Quadrature space.
Returns:: One array per cell.
Return type:: list[np.ndarray]

extract_values(function: dolfinx.fem.Function) → ndarray[source]

Extract all values from a Quadrature function.

Parameters:: function (dolfinx.fem.Function) – A function defined on the registered Quadrature space.
Returns:: Flat array of all integration-point values in cell order. Shape (total_points * dof_per_value,) or reshaped to (total_points, *tensor_shape) if tensor_shape is set.
Return type:: np.ndarray

property function_space: The Quadrature function space.

inject_cell_values(function: dolfinx.fem.Function, cell_index: int, values: ndarray) → None[source]

Inject values for a single cell.

Parameters:

function (dolfinx.fem.Function) – Target function on the Quadrature space.
cell_index (int) – The local cell index.
values (np.ndarray) – Values for the integration points of this cell.

inject_from_gather(function: dolfinx.fem.Function, cell_values: list[ndarray]) → None[source]

Inject per-cell values gathered from RVE solves.

Parameters:

function (dolfinx.fem.Function) – Target function on the Quadrature space.
cell_values (list[np.ndarray]) – One array per cell, as returned by the RVE workers.

inject_values(function: dolfinx.fem.Function, values: ndarray) → None[source]

Inject values into a Quadrature function.

Parameters:

function (dolfinx.fem.Function) – Target function on the Quadrature space.
values (np.ndarray) – Values to inject. Can be flat (total_points * dof_per_value,) or shaped (total_points, *tensor_shape).

property quadrature_data: QuadratureData: Metadata about the registered quadrature configuration.

register(mesh: dolfinx.mesh.Mesh, quadrature_degree: int = 2, tensor_shape: Tuple[int, ...] | int = ()) → QuadratureData[source]

Register a mesh and create a Quadrature function space.

Parameters:

mesh (dolfinx.mesh.Mesh) – The computational mesh (typically the macro-mesh in FE²).
quadrature_degree (int, optional) – The polynomial degree for the quadrature rule. Default = 2.
tensor_shape (tuple[int, ...] or int, optional) –
Shape of the tensor at each integration point. Examples:
- () or 1 — scalar field.
- (3,) — vector in 2D (Voigt stress/strain).
- (6,) — symmetric tensor in 3D (Voigt notation).
- (3, 3) — full 2D tensor.

Returns:

Metadata about the quadrature points.

Return type:

QuadratureData

register_with_function_space(function_space: dolfinx.fem.FunctionSpace, tensor_shape: Tuple[int, ...] | int = ()) → QuadratureData[source]

Register using a pre-existing Quadrature function space.

Parameters:

function_space (dolfinx.fem.FunctionSpace) – An existing Quadrature function space.
tensor_shape (tuple[int, ...] or int, optional) – Shape of the tensor at each integration point.

Return type:

QuadratureData

fenicsx_cosim.scatter_gather_communicator module

ScatterGatherCommunicator — Parallel RVE dispatch for FE² homogenization.

In a standard co-simulation, a zmq.PAIR socket connects exactly two solvers. For FE² however, a single macroscopic mesh might have 10 000+ integration points, each requiring an independent RVE solve. Running them sequentially is prohibitively slow.

The ScatterGatherCommunicator replaces the 1-to-1 topology with a PUSH/PULL fan-out / fan-in pattern:

Master (Macro-Solver): Pushes an array of strain tensors to the work queue via a PUSH socket. Gathers results from a PULL socket.
Workers (Micro-Solvers / RVEs): Each pulls a work item from the queue, solves the local RVE problem, and pushes the result back.

This corresponds to Section 1.3 of the Advanced Features Addendum.

Architecture diagram:

┌──────────┐  PUSH ──►  ┌────────────┐  PULL
│  Master  │ ──────────►│            │ ──────────►  Worker 0
│  (Macro) │ ──────────►│  Ventilator│ ──────────►  Worker 1
│          │ ──────────►│  (queue)   │ ──────────►  Worker 2
└──────────┘            └────────────┘             ...
     ▲                                              │
     │  PULL  ◄──  ┌────────────┐  PUSH ◄────────  │
     │ ◄────────── │  Collector │ ◄──────────────── │
     │ ◄────────── │  (gather)  │ ◄──────────────── │
                   └────────────┘

Typical usage (Master)

>>> sg = ScatterGatherCommunicator(role="master",
...     push_endpoint="tcp://*:5556",
...     pull_endpoint="tcp://*:5557")
>>> sg.scatter(work_items)          # list of np.ndarray
>>> results = sg.gather(n_expected) # list of np.ndarray

Typical usage (Worker)

>>> sg = ScatterGatherCommunicator(role="worker",
...     push_endpoint="tcp://master:5557",
...     pull_endpoint="tcp://master:5556")
>>> while True:
...     idx, data = sg.pull_work()
...     result = solve_rve(data)
...     sg.push_result(idx, result)

class fenicsx_cosim.scatter_gather_communicator.ScatterGatherCommunicator(role: Literal['master', 'worker'], push_endpoint: str = 'tcp://*:5556', pull_endpoint: str = 'tcp://*:5557', timeout_ms: int | None = None)[source]

Bases: object

Fan-out / fan-in communicator for dispatching RVE solves.

Parameters:

role ({"master", "worker"}) – Whether this process is the macro-solver (master) or an RVE worker.
push_endpoint (str) –
ZeroMQ endpoint for the PUSH socket.
- Master: binds this endpoint (ventilator).
- Worker: connects to this endpoint (to push results back).
pull_endpoint (str) –
ZeroMQ endpoint for the PULL socket.
- Master: binds this endpoint (collector).
- Worker: connects to this endpoint (to pull work items).
timeout_ms (int, optional) – Receive timeout for blocking pulls. Default 300 000 ms for master (waiting for RVE results), 60 000 ms for workers.

broadcast_shutdown(n_workers: int) → None[source]

Send shutdown signals to all workers.

Parameters:: n_workers (int) – Number of workers to notify.

close() → None[source]: Close all sockets.

gather(n_expected: int, ordered: bool = True) → list[ndarray][source]

Gather results from the worker pool.

Blocks until all n_expected results have been received.

Parameters:

n_expected (int) – Number of results to wait for.
ordered (bool, optional) – If True (default), results are returned in the same order as the scattered work items (by index). If False, results are returned in arrival order.

Returns:

The gathered results.

Return type:

list[np.ndarray]

Raises:

TimeoutError – If waiting for results exceeds timeout_ms.

property is_connected: bool: Whether the sockets are open.

pull_work() → tuple[int, ndarray, dict][source]

Pull a single work item from the master.

Returns:

(index, data_array, metadata_dict)

Return type:

tuple[int, np.ndarray, dict]

Raises:

StopIteration – If a shutdown signal is received.
TimeoutError – If no work arrives within timeout_ms.

push_result(index: int, result: ndarray, metadata: dict | None = None) → None[source]

Push a completed result back to the master.

Parameters:

index (int) – The work item index (received from pull_work()).
result (np.ndarray) – The computed result for this work item.
metadata (dict, optional) – Optional result metadata.

scatter(work_items: list[ndarray], metadata: list[dict] | None = None) → int[source]

Scatter work items to the worker pool.

Each work item is assigned a sequential index and sent as a multipart ZeroMQ message.

Parameters:

work_items (list[np.ndarray]) – One array per work unit (e.g. strain tensor per cell/RVE).
metadata (list[dict], optional) – Optional metadata dicts to send alongside each work item (e.g. material properties, damage state).

Returns:

The number of work items dispatched.

Return type:

int

scatter_gather(work_items: list[ndarray], metadata: list[dict] | None = None) → list[ndarray][source]

Convenience: scatter work items and then gather all results.

Parameters:

work_items (list[np.ndarray]) – Work items to dispatch.
metadata (list[dict], optional) – Optional per-item metadata.

Returns:

Results in the same order as the input work items.

Return type:

list[np.ndarray]

work_loop(solve_fn, **kwargs) → int[source]

Convenience: pull work items and push results until shutdown.

Parameters:

solve_fn (callable) – solve_fn(index: int, data: np.ndarray, metadata: dict) -> np.ndarray — the RVE solver function.
**kwargs – Additional keyword arguments passed to solve_fn.

Returns:

Number of work items processed before shutdown.

Return type:

int

fenicsx_cosim.utils module

Utility functions for fenicsx-cosim.

Provides logging configuration, serialization helpers for NumPy arrays over ZeroMQ, and common constants used across the package.

fenicsx_cosim.utils.deserialize_array(frames: list[bytes]) → tuple[str, Any][source]

Reconstruct a named array or matrix from a multipart ZeroMQ message.

Parameters:: frames (list[bytes]) – A multi-frame message as produced by serialize_array().
Returns:: (name, array) – the identifier and reconstructed data.
Return type:: tuple[str, Any]

fenicsx_cosim.utils.get_logger(name: str, level: int = 20) → Logger[source]

Create a consistently formatted logger for a cosim component.

Parameters:

name (str) – Logger name (typically the module __name__).
level (int, optional) – Logging level, by default logging.INFO.

Return type:

logging.Logger

fenicsx_cosim.utils.make_handshake_msg(my_name: str, partner_name: str) → bytes[source]

Build a handshake message.

Format: MAGIC | len(my_name) | my_name | len(partner_name) | partner_name

fenicsx_cosim.utils.parse_handshake_msg(data: bytes) → tuple[str, str][source]

Parse a handshake message.

Returns:: (sender_name, expected_partner_name)
Return type:: tuple[str, str]
Raises:: ValueError – If the magic bytes do not match.

fenicsx_cosim.utils.serialize_array(name: str, array: Any) → list[bytes][source]

Serialize a named NumPy array or SciPy sparse matrix into a multipart ZeroMQ message.

Parameters:

name (str) – A human-readable identifier for the data (e.g. "TemperatureField").
array (np.ndarray or scipy.sparse.spmatrix) – The data to transmit.

Returns:

A multi-frame message.

Return type:

list[bytes]

Module contents

fenicsx-cosim: A Native Partitioned Multiphysics Coupling Library for FEniCSx.

Inspired by the Kratos CoSimIO architecture, this package provides a non-intrusive API for connecting independent FEniCSx solvers across different processes for partitioned co-simulation.

Core modules

CouplingInterface — The main user API for boundary coupling, AMR, and FE² scatter-gather workflows.
MeshExtractor — Extracts boundary DoFs and coordinates from FEniCSx.
Communicator — ZeroMQ-based 1-to-1 inter-process communication.
DataMapper / NearestNeighborMapper — Non-conforming mesh interpolation.
DynamicMapper — AMR-aware mapping with automatic re-negotiation.
QuadratureExtractor — Integration-point data for FE² homogenization.
ScatterGatherCommunicator — PUSH/PULL fan-out for parallel RVE dispatch.

Example (Standard Boundary Coupling)

>>> from fenicsx_cosim import CouplingInterface
>>> cosim = CouplingInterface(name="ThermalSolver", partner_name="MechanicalSolver")
>>> cosim.register_interface(mesh, facet_tags, marker_id=1)
>>> cosim.export_data("TemperatureField", temperature_function)
>>> cosim.import_data("DisplacementField", displacement_function)

Example (FE² Scatter-Gather)

>>> cosim_fe2 = CouplingInterface(name="Macro", role="Master",
...     topology="scatter-gather")
>>> cosim_fe2.register_quadrature_space(V_quad)
>>> cosim_fe2.scatter_data("StrainTensor", macro_strains)
>>> stresses = cosim_fe2.gather_data("StressTensor")

class fenicsx_cosim.AbaqusFileAdapter(exchange_dir: str | Path, timeout_s: float = 60.0)[source]

Bases: SolverAdapter

Adapter for file-based coupling with Abaqus.

Parameters:

exchange_dir (str | Path) – Path to the shared directory where .npy files are written/read.
timeout_s (float, optional) – Maximum time to wait for expected files to appear. Currently unused as exact file presence is checked instantly for simplicity, but could be extended for polling file locks.

advance() → None[source]: No-op — File synchronization handles step progression implicitly.

extract_field(field_name: str) → ndarray[source]

Read field from ‘<field_name>_out.npy’.

Parameters:: field_name (str) – Name of the field.
Returns:: Shape (N,) or (N, 3).
Return type:: np.ndarray

get_boundary_coordinates() → ndarray[source]

Read coordinates from ‘boundary_coords.npy’.

Returns:: Shape (N, 3).
Return type:: np.ndarray
Raises:: FileNotFoundError – If the coordinates file does not exist.

get_metadata() → dict[str, str][source]

Return solver metadata (mesh info, solver name, etc.).

Override to provide metadata that the partner solver can inspect.

inject_field(field_name: str, values: ndarray) → None[source]

Write field to ‘<field_name>_in.npy’.

Parameters:

field_name (str) – Name of the field.
values (np.ndarray) – Values to write.

class fenicsx_cosim.Communicator(name: str, partner_name: str, role: Literal['bind', 'connect'], endpoint: str = 'tcp://localhost:5555', timeout_ms: int = 30000, handshake: bool = True)[source]

Bases: object

ZeroMQ-based communicator for inter-process data exchange.

Two coupled solvers each create a Communicator. One must take the "bind" role (the server) and the other the "connect" role (the client). The underlying socket type is zmq.PAIR which gives bidirectional, exclusive communication — ideal for a two-solver coupling scenario.

Parameters:

name (str) – Identifier for this solver.
partner_name (str) – Expected identifier of the partner solver.
role ({"bind", "connect"}) – Whether this communicator binds (acts as server) or connects (acts as client).
endpoint (str, optional) – A ZeroMQ endpoint string. Defaults to "tcp://localhost:5555". For same-machine IPC use "ipc:///tmp/fenicsx_cosim".
timeout_ms (int, optional) – Timeout in milliseconds for blocking receives. Default 30 000.
handshake (bool, optional) – If True (default), perform a name-exchange handshake upon initialization to verify that both sides are correctly paired.

close() → None[source]: Gracefully close the ZeroMQ socket.

property is_connected: bool: Whether the communicator socket is still open.

receive_array() → tuple[str, np.ndarray | 'sps.csr_matrix'][source]

Receive a named NumPy array from the partner.

Returns:: (data_name, array)
Return type:: tuple[str, np.ndarray | sps.csr_matrix]
Raises:: CommunicationError – If a timeout occurs or the socket is closed.

receive_raw() → bytes[source]: Receive raw bytes from the partner.

send_array(data_name: str, array: np.ndarray | 'sps.csr_matrix') → None[source]

Serialize and send a named NumPy array to the partner.

Parameters:

data_name (str) – Identifier for the data (e.g. "TemperatureField").
array (np.ndarray or scipy.sparse.spmatrix) – The data to transmit.

Raises:

CommunicationError – If the socket is closed or the send fails.

send_raw(data: bytes) → None[source]: Send raw bytes to the partner.

synchronize() → None[source]

Block until both solvers reach this synchronization point.

The bind side sends the sync signal first, then waits for ACK. The connect side waits for the sync signal, then sends ACK.

class fenicsx_cosim.CouplingInterface(name: str, partner_name: str = 'Workers', role: str | None = None, endpoint: str | None = None, connection_type: str = 'tcp', timeout_ms: int = 30000, enable_mapping: bool = True, topology: str = 'pair', push_endpoint: str = 'tcp://*:5556', pull_endpoint: str = 'tcp://*:5557')[source]

Bases: object

High-level co-simulation coupling API for FEniCSx solvers.

This is the single entry point that researchers interact with. It orchestrates boundary-data extraction, inter-process communication, and (optionally) non-conforming mesh mapping.

Parameters:

name (str) – A unique identifier for this solver instance (e.g. "ThermalSolver").
partner_name (str) – The identifier of the partner solver. For scatter-gather topology, this can be set to "Workers".
role ({"bind", "connect", "Master", "Worker"}, optional) – ZeroMQ role. The solver that starts first should "bind"; the one that starts second should "connect". If None (default), the role is determined automatically by sorting the names lexicographically — the name that comes first binds. Use "Master" / "Worker" for scatter-gather topology.
endpoint (str, optional) – ZeroMQ endpoint. Default "tcp://localhost:5555".
connection_type ({"tcp", "ipc"}, optional) – Shortcut to auto-generate an endpoint. Ignored if endpoint is explicitly provided.
timeout_ms (int, optional) – Receive timeout in ms (default 30 000).
enable_mapping (bool, optional) – If True (default), a NearestNeighborMapper is built automatically when both sides exchange boundary coordinates.
topology ({"pair", "scatter-gather"}, optional) – Communication topology. "pair" (default) uses the standard zmq.PAIR socket for 1-to-1 coupling. "scatter-gather" uses PUSH/PULL for FE² fan-out / fan-in.
push_endpoint (str, optional) – For scatter-gather only: PUSH socket endpoint.
pull_endpoint (str, optional) – For scatter-gather only: PULL socket endpoint.

advance_adapter() → None[source]: Synchronize and advance the adapter’s solver.

advance_in_time() → None[source]

Synchronize both solvers at the end of a time step.

This acts as a barrier — neither solver proceeds until both have called advance_in_time().

property boundary_coordinates: ndarray: Boundary DoF coordinates of this solver — shape (N, 3).

check_mesh_update() → bool[source]

Lightweight check to see if the partner solver updated its mesh.

If the partner calls update_interface_geometry(), this solver must call check_mesh_update() at the same point in the time loop to receive the new coordinates and re-build the mapping.

If neither solver refined its mesh but AMR is active, both solvers should call check_mesh_update() to acknowledge no changes.

Returns:: True if the partner updated its mesh and the mapping was rebuilt; False otherwise.
Return type:: bool

disconnect() → None[source]: Gracefully close the connection to the partner solver.

property dynamic_mapper: DynamicMapper | None: The dynamic (AMR-aware) mapper, if active.

export_data(data_name: str, function: dolfinx.fem.Function) → None[source]

Extract boundary values from a FEniCSx Function and send them to the partner solver.

Parameters:

data_name (str) – Identifier for the data (e.g. "TemperatureField").
function (dolfinx.fem.Function) – The FEniCSx function whose boundary values are to be sent.

export_raw(data_name: str, array: np.ndarray | 'sps.csr_matrix') → None[source]

Send a raw NumPy array or SciPy sparse matrix to the partner solver.

This is useful for sending non-field data like sparse stiffness matrices or scalar values (e.g. for Shakedown analysis or custom optimization loops).

Parameters:

data_name (str) – Identifier for the data.
array (np.ndarray or scipy.sparse.spmatrix) – The raw data to transmit.

export_via_adapter(field_name: str) → None[source]

Extract field from adapter and send to partner.

Parameters:: field_name (str) – Field name as understood by the adapter.

property extractor: MeshExtractor: The underlying mesh extractor.

classmethod from_adapter(adapter: SolverAdapter, name: str, partner_name: str = 'Partner', role: str | None = None, endpoint: str | None = None, connection_type: str = 'tcp', timeout_ms: int = 30000, enable_mapping: bool = True) → CouplingInterface[source]

Create a CouplingInterface using any SolverAdapter.

This allows non-FEniCSx solvers (Kratos, Abaqus, etc.) to participate in fenicsx-cosim coupling via their respective adapters.

Parameters:

adapter (SolverAdapter) – A concrete adapter for the solver.
name (str) – Identifier for this solver.
partner_name (str, optional) – Identifier of the partner solver.
role ({"bind", "connect"}, optional) – ZeroMQ role. Auto-determined if None.
endpoint (str, optional) – ZeroMQ endpoint.
connection_type ({"tcp", "ipc"}, optional) – Transport type.
timeout_ms (int, optional) – Receive timeout.
enable_mapping (bool, optional) – Build nearest-neighbour mapper.

Return type:

CouplingInterface

gather_data(data_name: str, function: dolfinx.fem.Function | None = None, n_expected: int | None = None) → list[ndarray][source]

Gather results from the RVE worker pool and optionally inject them into a Quadrature function.

Parameters:

data_name (str) – Name identifier (e.g. "StressTensor").
function (dolfinx.fem.Function, optional) – If provided, the gathered values are injected into this Quadrature function.
n_expected (int, optional) – Number of results to wait for. Defaults to quadrature_data.num_cells.

Returns:

The gathered per-cell result arrays.

Return type:

list[np.ndarray]

import_data(data_name: str, function: dolfinx.fem.Function) → None[source]

Receive boundary values from the partner solver and inject them into a FEniCSx Function.

If the meshes are non-conforming (different DoF coordinates), the received values are first mapped using the DataMapper before injection.

Parameters:

data_name (str) – Expected identifier for the incoming data.
function (dolfinx.fem.Function) – The target FEniCSx function.

import_raw(data_name: str) → np.ndarray | 'sps.csr_matrix'[source]

Receive a raw NumPy array or SciPy sparse matrix from the partner solver.

Parameters:: data_name (str) – Expected identifier for the incoming data.
Returns:: The raw data received.
Return type:: np.ndarray or scipy.sparse.spmatrix

import_via_adapter(field_name: str) → None[source]

Receive field from partner and inject into adapter.

Parameters:: field_name (str) – Expected field name.

property mapper: DataMapper | None: The data mapper (if mapping is enabled and built).

property partner_coordinates: ndarray | None: Boundary DoF coordinates received from the partner.

property quadrature_extractor: QuadratureExtractor | None: The quadrature extractor, if registered.

register_adapter_interface() → None[source]

Register the coupling interface using the attached adapter.

Extracts boundary coordinates from the adapter and exchanges them with the partner solver to build the data mapping.

Raises:: RuntimeError – If no adapter has been set (use from_adapter()).

register_interface(mesh: dolfinx.mesh.Mesh, facet_tags: dolfinx.mesh.MeshTags, marker_id: int, function_space: dolfinx.fem.FunctionSpace | None = None) → None[source]

Register the coupling boundary and exchange coordinates with the partner solver.

Parameters:

mesh (dolfinx.mesh.Mesh) – The computational mesh.
facet_tags (dolfinx.mesh.MeshTags) – Mesh tags labelling boundary facets.
marker_id (int) – The integer marker for the coupling boundary.
function_space (dolfinx.fem.FunctionSpace, optional) – The function space. If None, a default scalar Lagrange-1 space is created on mesh.

register_interface_from_locator(mesh: dolfinx.mesh.Mesh, locator_fn, function_space: dolfinx.fem.FunctionSpace | None = None, marker_id: int = 0) → None[source]

Register the coupling boundary using a geometric locator function.

Parameters:

mesh (dolfinx.mesh.Mesh) – The computational mesh.
locator_fn (callable) – locator_fn(x: ndarray) -> ndarray[bool]
function_space (dolfinx.fem.FunctionSpace, optional) – Defaults to Lagrange-1.
marker_id (int, optional) – Label for this boundary.

register_quadrature_space(function_space_or_mesh=None, quadrature_degree: int = 2, tensor_shape=()) → None[source]

Register a Quadrature function space for FE² homogenization.

This replaces the standard boundary registration and instead sets up data exchange at integration points.

Parameters:

function_space_or_mesh (dolfinx.fem.FunctionSpace or dolfinx.mesh.Mesh) – Either a pre-existing Quadrature function space, or a mesh on which to create one.
quadrature_degree (int, optional) – If a mesh is provided, the quadrature degree. Default 2.
tensor_shape (tuple[int, ...], optional) – Shape of the tensor at each integration point.

scatter_data(data_name: str, function: dolfinx.fem.Function, metadata: list[dict] | None = None) → int[source]

Extract per-cell integration-point values and scatter them to the RVE worker pool.

Parameters:

data_name (str) – Name identifier (e.g. "StrainTensor").
function (dolfinx.fem.Function) – Source function on the Quadrature space.
metadata (list[dict], optional) – Optional per-cell metadata (material params, damage state).

Returns:

Number of work items dispatched.

Return type:

int

scatter_gather_data(scatter_name: str, scatter_function: dolfinx.fem.Function, gather_name: str, gather_function: dolfinx.fem.Function | None = None, metadata: list[dict] | None = None) → list[ndarray][source]

Convenience: scatter, then gather in one call.

Parameters:

scatter_name (str) – Name for the outgoing data.
scatter_function (dolfinx.fem.Function) – Source Quadrature function.
gather_name (str) – Name for the incoming data.
gather_function (dolfinx.fem.Function, optional) – Target Quadrature function for injection.
metadata (list[dict], optional) – Per-cell metadata.

Returns:

Gathered results.

Return type:

list[np.ndarray]

property step_count: int: Number of completed time steps.

update_interface_geometry(mesh: dolfinx.mesh.Mesh, facet_tags: dolfinx.mesh.MeshTags, marker_id: int, function_space: dolfinx.fem.FunctionSpace | None = None) → None[source]

Notify the coupling system that the mesh has been refined (AMR).

This method:

Re-extracts boundary DoFs from the refined mesh.
Invalidates the current data mapping.
Negotiates a synchronized coordinate exchange with the partner solver.
Rebuilds the DynamicMapper with the updated point clouds.

Parameters:

mesh (dolfinx.mesh.Mesh) – The newly refined mesh.
facet_tags (dolfinx.mesh.MeshTags) – Updated facet tags for the refined mesh.
marker_id (int) – The coupling boundary marker.
function_space (dolfinx.fem.FunctionSpace, optional) – Updated function space (if None, a new Lagrange-1 space is created on the refined mesh).

See also

DynamicMapper.negotiate_update

class fenicsx_cosim.DataMapper[source]

Bases: ABC

Abstract base class for all data mapping strategies.

Subclasses must implement build() and map().

abstractmethod build(source_coords: ndarray, target_coords: ndarray) → None[source]

Build the mapping from source to target point clouds.

Parameters:

source_coords (np.ndarray) – Coordinates of the source DoFs, shape (N_src, 3).
target_coords (np.ndarray) – Coordinates of the target DoFs, shape (N_tgt, 3).

abstractmethod inverse_map(target_values: ndarray) → ndarray[source]

Map values from target DoFs back to source DoFs.

Parameters:: target_values (np.ndarray) – Field values at the target DoFs.
Returns:: Interpolated values at the source DoFs.
Return type:: np.ndarray

abstractmethod map(source_values: ndarray) → ndarray[source]

Map values from source DoFs to target DoFs.

Parameters:: source_values (np.ndarray) – Field values at the source DoFs, shape (N_src,) or (N_src, D) for vector fields.
Returns:: Interpolated values at the target DoFs.
Return type:: np.ndarray

class fenicsx_cosim.DynamicMapper(base_mapper_cls: type[DataMapper] = <class 'fenicsx_cosim.data_mapper.NearestNeighborMapper'>)[source]

Bases: object

Mesh-change–aware wrapper around DataMapper.

The DynamicMapper keeps track of a revision counter which is bumped every time the mesh changes. It provides helpers to negotiate a synchronized re-mapping with the partner solver.

Parameters:: base_mapper_cls (type[DataMapper], optional) – The concrete mapper to use. Defaults to NearestNeighborMapper.

revision

The number of times the mapping has been (re-)built.

Type:: int

max_distance

Delegates to the underlying DataMapper.max_distance.

Type:: float or None

build(source_coords: ndarray, target_coords: ndarray) → None[source]

Build (or rebuild) the underlying mapping.

Parameters:

source_coords (np.ndarray) – Partner (source) boundary coordinates — shape (N_src, 3).
target_coords (np.ndarray) – Local (target) boundary coordinates — shape (N_tgt, 3).

invalidate() → None[source]

Mark the current mapping as stale.

This should be called whenever the local mesh has been refined or changed. The mapping will not be usable until build() is called again.

inverse_map(target_values: ndarray) → ndarray[source]: Map values from local (target) back to partner (source) DoFs.

property local_coordinates: ndarray | None: The local boundary coordinates used in the latest mapping.

map(source_values: ndarray) → ndarray[source]

Map values from partner (source) to local (target) DoFs.

Raises RuntimeError if the mapping is stale.

property max_distance: float | None: Maximum mapping distance from the underlying mapper.

property needs_update: bool: True if the mapping has been invalidated and must be rebuilt.

negotiate_update(communicator, role: str, my_new_coords: ndarray | None) → ndarray | None[source]

Negotiate a mesh-update with the partner solver.

Both solvers must call this method at the same point in the time loop. The protocol:

The bind side sends either UPDATE_MESH_SIGNAL or NO_UPDATE_SIGNAL.
The connect side receives the signal and then sends its own signal (likewise UPDATE_MESH_SIGNAL or NO_UPDATE_SIGNAL).
If either side signals an update, both exchange new boundary coordinates and return the partner’s new coordinates. Otherwise None is returned and the mapping is untouched.

Parameters:

communicator (Communicator) – The ZeroMQ communicator already connected to the partner.
role (str) – "bind" or "connect".
my_new_coords (np.ndarray or None) – The new local boundary coordinates if the mesh changed, or None if no change occurred.

Returns:

The partner’s new boundary coordinates if an exchange took place, or None if no update was needed.

Return type:

np.ndarray or None

property partner_coordinates: ndarray | None: The partner boundary coordinates used in the latest mapping.

class fenicsx_cosim.KratosAdapter(model_part: KratosMultiphysics.ModelPart, interface_sub_model_part_name: str, variable_map: dict[str, Any] | None = None)[source]

Bases: SolverAdapter

Adapter for coupling Kratos Multiphysics solvers via fenicsx-cosim.

Parameters:

model_part (KratosMultiphysics.ModelPart) – The main Kratos model part containing the simulation data.
interface_sub_model_part_name (str) – Name of the sub-model-part that defines the coupling boundary. All coordinate/field extraction happens on this sub-model-part’s nodes.
variable_map (dict[str, Any], optional) – Mapping from string field names to Kratos variable objects. If None (default), variables are looked up dynamically via KratosMultiphysics.KratosGlobals.GetVariable(field_name).

Examples

>>> adapter = KratosAdapter(model_part, "interface_boundary")
>>> coords = adapter.get_boundary_coordinates()  # (N, 3) array
>>> temp = adapter.extract_field("TEMPERATURE")   # (N,) array
>>> adapter.inject_field("DISPLACEMENT_X", disp_x) # set values

advance() → None[source]: No-op — Kratos time-stepping is managed by the user script.

extract_field(field_name: str) → ndarray[source]

Extract a scalar field from interface nodes.

Parameters:: field_name (str) – Kratos variable name (e.g. "TEMPERATURE", "DISPLACEMENT_X").
Returns:: Shape (N,) — one value per interface node.
Return type:: np.ndarray

extract_vector_field(field_name: str) → ndarray[source]

Extract a vector field (e.g. DISPLACEMENT, VELOCITY).

Parameters:: field_name (str) – Kratos vector variable name.
Returns:: Shape (N, 3) — (x, y, z) components per node.
Return type:: np.ndarray

get_boundary_coordinates() → ndarray[source]

Return coupling boundary node coordinates.

Returns:: Shape (N, 3) — (x, y, z) for each interface node, in the same order as node_ids.
Return type:: np.ndarray

get_field_names() → list[str][source]: Return user-registered field names.

get_metadata() → dict[str, Any][source]: Return summary metadata about the Kratos interface.

inject_field(field_name: str, values: ndarray) → None[source]

Set a scalar field at interface nodes.

Parameters:

field_name (str) – Kratos variable name.
values (np.ndarray) – Shape (N,) — one value per interface node.

Raises:

ValueError – If values length doesn’t match the number of interface nodes.

inject_vector_field(field_name: str, values: ndarray) → None[source]

Set a vector field at interface nodes.

Parameters:

field_name (str) – Kratos vector variable name.
values (np.ndarray) – Shape (N, 3).

property node_ids: list[int]: Interface node IDs in consistent order.

property num_nodes: int: Number of coupling boundary nodes.

class fenicsx_cosim.MeshExtractor[source]

Bases: object

Extract boundary DoFs and coordinates from a FEniCSx mesh.

This class wraps the FEniCSx v0.10+ API to:

Identify boundary facets using mesh-tags or a locator function.
Find the DoFs living on those facets via topological DoF location.
Extract the (x, y, z) coordinates of the boundary DoFs.

The extracted data is stored internally and can be queried as NumPy arrays for serialization and transmission via the Communicator.

property boundary_coordinates: ndarray: Spatial coordinates of boundary DoFs — shape (N, 3).

property boundary_data: BoundaryData: The most recently extracted BoundaryData.

property boundary_dof_indices: ndarray: Global DoF indices on the coupling boundary — shape (N,).

extract_boundary_values(function: dolfinx.fem.Function) → ndarray[source]

Extract the values of a FEniCSx Function at the boundary DoFs.

Parameters:: function (dolfinx.fem.Function) – The FEniCSx function whose boundary values are needed.
Returns:: Values at the boundary DoFs.
Return type:: np.ndarray

inject_boundary_values(function: dolfinx.fem.Function, values: ndarray) → None[source]

Inject values into a FEniCSx Function at the boundary DoFs.

Parameters:

function (dolfinx.fem.Function) – The target FEniCSx function.
values (np.ndarray) – The values to inject (must match the number of boundary DoFs).

Raises:

ValueError – If values length does not match the number of boundary DoFs.

register(mesh: dolfinx.mesh.Mesh, facet_tags: dolfinx.mesh.MeshTags, marker_id: int, function_space: dolfinx.fem.FunctionSpace) → BoundaryData[source]

Register a coupling boundary and extract its DoF information.

Parameters:

mesh (dolfinx.mesh.Mesh) – The computational mesh.
facet_tags (dolfinx.mesh.MeshTags) – Mesh tags that label boundary facets.
marker_id (int) – The integer marker identifying the coupling boundary within facet_tags.
function_space (dolfinx.fem.FunctionSpace) – The function space whose DoFs on the boundary are needed.

Returns:

A dataclass holding the extracted boundary information.

Return type:

BoundaryData

register_from_locator(mesh: dolfinx.mesh.Mesh, locator_fn, function_space: dolfinx.fem.FunctionSpace, marker_id: int = 0) → BoundaryData[source]

Register a coupling boundary using a geometric locator function.

This is a convenience method for cases where no MeshTags object exists. The locator_fn must accept an (3, N) array of coordinates and return a boolean mask of length N.

Parameters:

mesh (dolfinx.mesh.Mesh) – The computational mesh.
locator_fn (callable) – locator_fn(x: ndarray) -> ndarray[bool] — identifies boundary facets.
function_space (dolfinx.fem.FunctionSpace) – The function space whose DoFs on the boundary are needed.
marker_id (int, optional) – A label for this boundary (default 0).

Return type:

BoundaryData

class fenicsx_cosim.NearestNeighborMapper[source]

Bases: DataMapper

Maps data by assigning each target DoF the value of its nearest source DoF.

This is the simplest (and often most robust) mapping strategy. It is suitable when the two meshes have similar resolution on the coupling boundary and the fields vary smoothly.

max_distance

After building, the maximum distance between any target point and its nearest source point. Useful for diagnosing mesh mismatch issues.

Type:: float or None

build(source_coords: ndarray, target_coords: ndarray) → None[source]

Build the nearest-neighbor mapping.

Parameters:

source_coords (np.ndarray) – Shape (N_src, 3).
target_coords (np.ndarray) – Shape (N_tgt, 3).

property forward_distances: ndarray: Distances between each target point and its nearest source point.

inverse_map(target_values: ndarray) → ndarray[source]

Map values from target DoFs back to source DoFs.

Parameters:: target_values (np.ndarray) – Shape (N_tgt,) or (N_tgt, D).
Returns:: Shape (N_src,) or (N_src, D).
Return type:: np.ndarray

map(source_values: ndarray) → ndarray[source]

Map values from source DoFs to target DoFs using nearest neighbor.

Parameters:: source_values (np.ndarray) – Shape (N_src,) for scalar or (N_src, D) for vector fields.
Returns:: Shape (N_tgt,) or (N_tgt, D).
Return type:: np.ndarray

class fenicsx_cosim.QuadratureExtractor[source]

Bases: object

Extract and inject tensor data at integration points.

This class works with FEniCSx Quadrature function spaces to handle the data exchange required for FE² (computational homogenization).

The user creates a QuadratureExtractor, registers it with a mesh and quadrature degree, and then uses it to extract/inject tensors that are stored as dolfinx.fem.Function objects on the Quadrature space.

property cell_to_dof_map: ndarray: Cell → global DoF index mapping — shape (num_cells, dofs_per_cell).

extract_cell_values(function: dolfinx.fem.Function, cell_index: int) → ndarray[source]

Extract values for a single cell (element).

Parameters:

function (dolfinx.fem.Function) – A function on the Quadrature space.
cell_index (int) – The local cell index.

Returns:

Values at the integration points of the specified cell. Shape (points_per_cell,) for scalar, or (points_per_cell, *tensor_shape) for tensors.

Return type:

np.ndarray

extract_for_dispatch(function: dolfinx.fem.Function) → list[ndarray][source]

Extract per-cell integration-point values for dispatching to RVEs.

Returns a list of num_cells arrays, each of shape (points_per_cell,) or (points_per_cell, *tensor_shape). Cell i corresponds to RVE i.

Parameters:: function (dolfinx.fem.Function) – Source function on the Quadrature space.
Returns:: One array per cell.
Return type:: list[np.ndarray]

extract_values(function: dolfinx.fem.Function) → ndarray[source]

Extract all values from a Quadrature function.

Parameters:: function (dolfinx.fem.Function) – A function defined on the registered Quadrature space.
Returns:: Flat array of all integration-point values in cell order. Shape (total_points * dof_per_value,) or reshaped to (total_points, *tensor_shape) if tensor_shape is set.
Return type:: np.ndarray

property function_space: The Quadrature function space.

inject_cell_values(function: dolfinx.fem.Function, cell_index: int, values: ndarray) → None[source]

Inject values for a single cell.

Parameters:

function (dolfinx.fem.Function) – Target function on the Quadrature space.
cell_index (int) – The local cell index.
values (np.ndarray) – Values for the integration points of this cell.

inject_from_gather(function: dolfinx.fem.Function, cell_values: list[ndarray]) → None[source]

Inject per-cell values gathered from RVE solves.

Parameters:

function (dolfinx.fem.Function) – Target function on the Quadrature space.
cell_values (list[np.ndarray]) – One array per cell, as returned by the RVE workers.

inject_values(function: dolfinx.fem.Function, values: ndarray) → None[source]

Inject values into a Quadrature function.

Parameters:

function (dolfinx.fem.Function) – Target function on the Quadrature space.
values (np.ndarray) – Values to inject. Can be flat (total_points * dof_per_value,) or shaped (total_points, *tensor_shape).

property quadrature_data: QuadratureData: Metadata about the registered quadrature configuration.

register(mesh: dolfinx.mesh.Mesh, quadrature_degree: int = 2, tensor_shape: Tuple[int, ...] | int = ()) → QuadratureData[source]

Register a mesh and create a Quadrature function space.

Parameters:

mesh (dolfinx.mesh.Mesh) – The computational mesh (typically the macro-mesh in FE²).
quadrature_degree (int, optional) – The polynomial degree for the quadrature rule. Default = 2.
tensor_shape (tuple[int, ...] or int, optional) –
Shape of the tensor at each integration point. Examples:
- () or 1 — scalar field.
- (3,) — vector in 2D (Voigt stress/strain).
- (6,) — symmetric tensor in 3D (Voigt notation).
- (3, 3) — full 2D tensor.

Returns:

Metadata about the quadrature points.

Return type:

QuadratureData

register_with_function_space(function_space: dolfinx.fem.FunctionSpace, tensor_shape: Tuple[int, ...] | int = ()) → QuadratureData[source]

Register using a pre-existing Quadrature function space.

Parameters:

function_space (dolfinx.fem.FunctionSpace) – An existing Quadrature function space.
tensor_shape (tuple[int, ...] or int, optional) – Shape of the tensor at each integration point.

Return type:

QuadratureData

class fenicsx_cosim.ScatterGatherCommunicator(role: Literal['master', 'worker'], push_endpoint: str = 'tcp://*:5556', pull_endpoint: str = 'tcp://*:5557', timeout_ms: int | None = None)[source]

Bases: object

Fan-out / fan-in communicator for dispatching RVE solves.

Parameters:

role ({"master", "worker"}) – Whether this process is the macro-solver (master) or an RVE worker.
push_endpoint (str) –
ZeroMQ endpoint for the PUSH socket.
- Master: binds this endpoint (ventilator).
- Worker: connects to this endpoint (to push results back).
pull_endpoint (str) –
ZeroMQ endpoint for the PULL socket.
- Master: binds this endpoint (collector).
- Worker: connects to this endpoint (to pull work items).
timeout_ms (int, optional) – Receive timeout for blocking pulls. Default 300 000 ms for master (waiting for RVE results), 60 000 ms for workers.

broadcast_shutdown(n_workers: int) → None[source]

Send shutdown signals to all workers.

Parameters:: n_workers (int) – Number of workers to notify.

close() → None[source]: Close all sockets.

gather(n_expected: int, ordered: bool = True) → list[ndarray][source]

Gather results from the worker pool.

Blocks until all n_expected results have been received.

Parameters:

n_expected (int) – Number of results to wait for.
ordered (bool, optional) – If True (default), results are returned in the same order as the scattered work items (by index). If False, results are returned in arrival order.

Returns:

The gathered results.

Return type:

list[np.ndarray]

Raises:

TimeoutError – If waiting for results exceeds timeout_ms.

property is_connected: bool: Whether the sockets are open.

pull_work() → tuple[int, ndarray, dict][source]

Pull a single work item from the master.

Returns:

(index, data_array, metadata_dict)

Return type:

tuple[int, np.ndarray, dict]

Raises:

StopIteration – If a shutdown signal is received.
TimeoutError – If no work arrives within timeout_ms.

push_result(index: int, result: ndarray, metadata: dict | None = None) → None[source]

Push a completed result back to the master.

Parameters:

index (int) – The work item index (received from pull_work()).
result (np.ndarray) – The computed result for this work item.
metadata (dict, optional) – Optional result metadata.

scatter(work_items: list[ndarray], metadata: list[dict] | None = None) → int[source]

Scatter work items to the worker pool.

Each work item is assigned a sequential index and sent as a multipart ZeroMQ message.

Parameters:

work_items (list[np.ndarray]) – One array per work unit (e.g. strain tensor per cell/RVE).
metadata (list[dict], optional) – Optional metadata dicts to send alongside each work item (e.g. material properties, damage state).

Returns:

The number of work items dispatched.

Return type:

int

scatter_gather(work_items: list[ndarray], metadata: list[dict] | None = None) → list[ndarray][source]

Convenience: scatter work items and then gather all results.

Parameters:

work_items (list[np.ndarray]) – Work items to dispatch.
metadata (list[dict], optional) – Optional per-item metadata.

Returns:

Results in the same order as the input work items.

Return type:

list[np.ndarray]

work_loop(solve_fn, **kwargs) → int[source]

Convenience: pull work items and push results until shutdown.

Parameters:

solve_fn (callable) – solve_fn(index: int, data: np.ndarray, metadata: dict) -> np.ndarray — the RVE solver function.
**kwargs – Additional keyword arguments passed to solve_fn.

Returns:

Number of work items processed before shutdown.

Return type:

int

class fenicsx_cosim.SolverAdapter[source]

Bases: ABC

Bridge between any solver’s native data structures and fenicsx-cosim.

Subclasses must implement the four abstract methods below. The adapter is then passed to CouplingInterface.from_adapter() (or used standalone with a raw Communicator).

Design principles

No FEniCSx dependency — adapters for external solvers must not import dolfinx. Only the FEniCSxAdapter needs DOLFINx.
NumPy is the lingua franca — all data is exchanged as NumPy arrays (coordinates as (N, 3) float64, scalar fields as (N,) float64, vector fields as (N, dim) float64).
Thin wrappers — adapters should delegate to solver APIs, not re-implement solver logic.

abstractmethod advance() → None[source]

Advance the solver by one time step.

This is called by CouplingInterface.advance_in_time() after data exchange is complete. Implement as a no-op if the solver’s time-stepping is managed externally.

abstractmethod extract_field(field_name: str) → ndarray[source]

Extract a field’s values at coupling boundary nodes.

Parameters:: field_name (str) – Solver-native field identifier (e.g. "TEMPERATURE").
Returns:: Shape (N,) for scalar or (N, dim) for vector fields.
Return type:: np.ndarray

abstractmethod get_boundary_coordinates() → ndarray[source]

Return coordinates of coupling boundary nodes.

Returns:: Shape (N, 3), dtype float64. Each row is (x, y, z) for a boundary node.
Return type:: np.ndarray

get_field_names() → list[str][source]

Return a list of available field names for export.

Override to advertise which fields the solver can provide. The default returns an empty list (metadata exchange is optional).

get_metadata() → dict[str, Any][source]

Return solver metadata (mesh info, solver name, etc.).

Override to provide metadata that the partner solver can inspect.

abstractmethod inject_field(field_name: str, values: ndarray) → None[source]

Set field values at coupling boundary nodes.

Parameters:

field_name (str) – Solver-native field identifier.
values (np.ndarray) – Same shape convention as extract_field().