Coverage for /dolfinx-env/lib/python3.12/site-packages/io4dolfinx/backends/__init__.py: 88%
59 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-18 18:20 +0000
1from enum import Enum
2from importlib import import_module
3from pathlib import Path
4from typing import Any, Protocol
6from mpi4py import MPI
8import dolfinx
9import numpy as np
10import numpy.typing as npt
12from ..structures import ArrayData, FunctionData, MeshData, MeshTagsData, ReadMeshData
14__all__ = ["FileMode", "IOBackend", "get_backend", "set_default_backend"]
16_DEFAULT_BACKEND = "adios2"
19def set_default_backend(backend: str):
20 """Set the global default backend for io4dolfinx."""
21 global _DEFAULT_BACKEND
22 if backend not in BUILTIN_BAKENDS:
23 try:
24 get_backend(backend)
25 except ImportError as e:
26 raise ImportError(f"Backend {backend} not found.") from e
27 _DEFAULT_BACKEND = backend
30def get_default_backend():
31 """Get the global default backend for io4dolfinx."""
32 return _DEFAULT_BACKEND
35class ReadMode(Enum):
36 serial = 10 # This means that all data is read in on root rank
38 # Total number of data P, num processes = i + 1.
39 # All processes reads at least `P // (i+1)` items
40 # The first j=P%(i+1) processes reads `P // (i+1) + 1` items
41 # ```python
42 # def compute_partitioning(P, J):
43 # min_num = P // J
44 # num_per_proc = np.full(J, min_num)
45 # rem = P % J
46 # num_per_proc[:int(rem)] += 1
47 # assert(sum(num_per_proc)) == P
48 # return num_per_proc
49 # ```
50 parallel = 20
53class FileMode(Enum):
54 """Filen mode used for opening files."""
56 append = 10 #: Append data to file
57 write = 20 #: Write data to file
58 read = 30 #: Read data from file
61# See https://peps.python.org/pep-0544/#modules-as-implementations-of-protocols
62class IOBackend(Protocol):
63 read_mode: ReadMode
65 def get_default_backend_args(self, arguments: dict[str, Any] | None) -> dict[str, Any]:
66 """Get default backend arguments given a set of input arguments.
68 Args:
69 arguments: Input backend arguments
71 Returns:
72 Updated backend arguments
73 """
75 def write_attributes(
76 self,
77 filename: Path | str,
78 comm: MPI.Comm,
79 name: str,
80 attributes: dict[str, np.ndarray],
81 backend_args: dict[str, Any] | None,
82 ):
83 """Write attributes to file.
85 Args:
86 filename: Path to file to write to
87 comm: MPI communicator used in storage
88 name: Name of the attribute group
89 attributes: Dictionary of attributes to write
90 backend_args: Arguments to backend
91 """
93 def read_attributes(
94 self,
95 filename: Path | str,
96 comm: MPI.Comm,
97 name: str,
98 backend_args: dict[str, Any] | None,
99 ) -> dict[str, Any]:
100 """Read attributes from file.
102 Args:
103 filename: Path to file to read from
104 comm: MPI communicator used in storage
105 name: Name of the attribute group
106 backend_args: Arguments to backend
108 Returns:
109 Dictionary of attributes read from file
110 """
112 def read_timestamps(
113 self,
114 filename: Path | str,
115 comm: MPI.Comm,
116 function_name: str,
117 backend_args: dict[str, Any] | None,
118 ) -> npt.NDArray[np.float64 | str]: # type: ignore[type-var]
119 """Read timestamps from file.
121 Args:
122 filename: Path to file to read from
123 comm: MPI communicator used in storage
124 function_name: Name of the function to read timestamps for
125 backend_args: Arguments to backend
127 Returns:
128 Numpy array of timestamps read from file
129 """
131 def write_mesh(
132 self,
133 filename: Path | str,
134 comm: MPI.Comm,
135 mesh: MeshData,
136 backend_args: dict[str, Any] | None,
137 mode: FileMode,
138 time: float,
139 ):
140 """
141 Write a mesh to file.
143 Args:
144 comm: MPI communicator used in storage
145 mesh: Internal data structure for the mesh data to save to file
146 filename: Path to file to write to
147 backend_args: Arguments to backend
148 mode: File-mode to store the mesh
149 time: Time stamp associated with the mesh
150 """
152 def write_meshtags(
153 self,
154 filename: str | Path,
155 comm: MPI.Comm,
156 data: MeshTagsData,
157 backend_args: dict[str, Any] | None,
158 ):
159 """Write mesh tags to file.
161 Args:
162 filename: Path to file to write to
163 comm: MPI communicator used in storage
164 data: Internal data structure for the mesh tags to save to file
165 backend_args: Arguments to backend
166 """
168 def read_mesh_data(
169 self,
170 filename: Path | str,
171 comm: MPI.Comm,
172 time: str | float | None,
173 read_from_partition: bool,
174 backend_args: dict[str, Any] | None,
175 ) -> ReadMeshData:
176 """Read mesh data from file.
178 Args:
179 filename: Path to file to read from
180 comm: MPI communicator used in storage
181 time: Time stamp associated with the mesh to read
182 read_from_partition: Whether to read partition information
183 backend_args: Arguments to backend
185 Returns:
186 Internal data structure for the mesh data read from file
187 """
189 def read_meshtags_data(
190 self,
191 filename: str | Path,
192 comm: MPI.Comm,
193 name: str,
194 backend_args: dict[str, Any] | None,
195 ) -> MeshTagsData:
196 """Read mesh tags from file.
198 Args:
199 filename: Path to file to read from
200 comm: MPI communicator used in storage
201 name: Name of the mesh tags to read
202 backend_args: Arguments to backend
204 Returns:
205 Internal data structure for the mesh tags read from file
206 """
208 def read_dofmap(
209 self,
210 filename: str | Path,
211 comm: MPI.Comm,
212 name: str,
213 backend_args: dict[str, Any] | None,
214 ) -> dolfinx.graph.AdjacencyList:
215 """Read the dofmap of a function with a given name.
217 Args:
218 filename: Path to file to read from
219 comm: MPI communicator used in storage
220 name: Name of the function to read the dofmap for
221 backend_args: Arguments to backend
223 Returns:
224 Dofmap as an {py:class}`dolfinx.graph.AdjacencyList`
225 """
227 def read_dofs(
228 self,
229 filename: str | Path,
230 comm: MPI.Comm,
231 name: str,
232 time: float,
233 backend_args: dict[str, Any] | None,
234 ) -> tuple[npt.NDArray[np.float32 | np.float64 | np.complex64 | np.complex128], int]:
235 """Read the dofs (values) of a function with a given name from a given timestep.
237 Args:
238 filename: Path to file to read from
239 comm: MPI communicator used in storage
240 name: Name of the function to read the dofs for
241 time: Time stamp associated with the function to read
242 backend_args: Arguments to backend
244 Returns:
245 Contiguous sequence of degrees of freedom (with respect to input data)
246 and the global starting point on the process.
247 Process 0 has [0, M), process 1 [M, N), process 2 [N, O) etc.
248 """
250 def read_cell_perms(
251 self, comm: MPI.Comm, filename: Path | str, backend_args: dict[str, Any] | None
252 ) -> npt.NDArray[np.uint32]:
253 """
254 Read cell permutation from file with given communicator,
255 Split in continuous chunks based on number of cells in the input data.
257 Args:
258 comm: MPI communicator used in storage
259 filename: Path to file to read from
260 backend_args: Arguments to backend
262 Returns:
263 Contiguous sequence of permutations (with respect to input data)
264 Process 0 has [0, M), process 1 [M, N), process 2 [N, O) etc.
265 """
267 def write_function(
268 self,
269 filename: Path,
270 comm: MPI.Comm,
271 u: FunctionData,
272 time: float,
273 mode: FileMode,
274 backend_args: dict[str, Any] | None,
275 ):
276 """Write a function to file.
278 Args:
279 comm: MPI communicator used in storage
280 u: Internal data structure for the function data to save to file
281 filename: Path to file to write to
282 time: Time stamp associated with function
283 mode: File-mode to store the function
284 backend_args: Arguments to backend
285 """
287 def read_legacy_mesh(
288 self, filename: Path | str, comm: MPI.Comm, group: str
289 ) -> tuple[npt.NDArray[np.int64], npt.NDArray[np.floating], str | None]:
290 """Read in the mesh topology, geometry and (optionally) cell type from a
291 legacy DOLFIN HDF5-file.
293 Args:
294 filename: Path to file to read from
295 comm: MPI communicator used in storage
296 group: Group in HDF5 file where mesh is stored
298 Returns:
299 Tuple containing:
300 - Topology as a (num_cells, num_vertices_per_cell) array of global vertex indices
301 - Geometry as a (num_vertices, geometric_dimension) array of vertex coordinates
302 - Cell type as a string (e.g. "tetrahedron") or None if not found
303 """
305 def snapshot_checkpoint(
306 self,
307 filename: Path | str,
308 mode: FileMode,
309 u: dolfinx.fem.Function,
310 backend_args: dict[str, Any] | None,
311 ):
312 """Create a snapshot checkpoint of a dolfinx function.
314 Args:
315 filename: Path to file to read from
316 mode: File-mode to store the function
317 u: dolfinx function to create a snapshot checkpoint for
318 backend_args: Arguments to backend
319 """
321 def read_hdf5_array(
322 self,
323 comm: MPI.Comm,
324 filename: Path | str,
325 group: str,
326 backend_args: dict[str, Any] | None,
327 ) -> tuple[np.ndarray, int]:
328 """Read an array from an HDF5 file.
330 Args:
331 comm: MPI communicator used in storage
332 filename: Path to file to read from
333 group: Group in HDF5 file where array is stored
334 backend_args: Arguments to backend
336 Returns:
337 Tuple containing:
338 - Numpy array read from file
339 - Global starting point on the process.
340 Process 0 has [0, M), process 1 [M, N), process 2 [N, O) etc.
341 """
343 def read_point_data(
344 self,
345 filename: Path | str,
346 name: str,
347 comm: MPI.Comm,
348 time: str | float | None,
349 backend_args: dict[str, Any] | None,
350 ) -> tuple[np.ndarray, int]:
351 """Read data from the nodes of a mesh.
353 Args:
354 filename: Path to file
355 name: Name of point data
356 comm: Communicator to launch IO on.
357 time: The time stamp
358 backend_args: The backend arguments
359 Returns:
360 Data local to process (contiguous, no mpi comm) and local start range
361 """
362 ...
364 def read_function_names(
365 self, filename: Path | str, comm: MPI.Comm, backend_args: dict[str, Any] | None
366 ) -> list[str]:
367 """Read all function names from a file.
369 Args:
370 filename: Path to file
371 comm: MPI communicator to launch IO on.
372 backend_args: Arguments to backend
374 Returns:
375 A list of function names.
376 """
377 ...
379 def read_cell_data(
380 self,
381 filename: Path | str,
382 name: str,
383 comm: MPI.Comm,
384 time: str | float | None,
385 backend_args: dict[str, Any] | None,
386 ) -> tuple[npt.NDArray[np.int64], np.ndarray]:
387 """Read data from the cells of a mesh.
389 Args:
390 filename: Path to file
391 name: Name of point data
392 comm: Communicator to launch IO on.
393 time: The time stamp
394 backend_args: The backend arguments
395 Returns:
396 A tuple (topology, dofs) where topology contains the
397 vertex indices of the cells, dofs the degrees of
398 freedom within that cell.
399 """
400 ...
402 def write_data(
403 self,
404 filename: Path | str,
405 array_data: ArrayData,
406 comm: MPI.Comm,
407 time: str | float | None,
408 mode: FileMode,
409 backend_args: dict[str, Any] | None,
410 ):
411 """Write a 2D-array to file.
414 Args:
415 filename: Path to file
416 array_data: Data to write to file.
417 comm: The MPI communicator to open the writer with.
418 time: The time stamp
419 mode: Append or write
420 backend_args: The backend arguments
421 """
422 ...
425def get_backend(backend: str | None = None) -> IOBackend:
426 """Get backend class from backend name.
428 Args:
429 backend: Name of the backend to get
431 Returns:
432 Backend class
433 """
434 if backend is None:
435 backend = _DEFAULT_BACKEND
437 if backend == "h5py":
438 from .h5py import backend as H5PYInterface
440 return H5PYInterface
441 elif backend == "adios2":
442 from .adios2 import backend as ADIOS2Interface
444 return ADIOS2Interface
445 elif backend == "pyvista":
446 from .pyvista import backend as PYVISTAInterface
448 return PYVISTAInterface
449 elif backend == "xdmf":
450 from .xdmf import backend as XDMFInterface
452 return XDMFInterface
453 elif backend == "vtkhdf":
454 from .vtkhdf import backend as VTKDHFInterface
456 return VTKDHFInterface
457 elif backend == "exodus":
458 from .exodus import backend as EXODUSInterface
460 return EXODUSInterface
461 else:
462 return import_module(backend)
465BUILTIN_BAKENDS = ("h5py", "adios2", "pyvista", "xdmf", "vtkhdf", "exodus")
468def list_builtin_backends() -> list[str]:
469 """List available builtin backends.
471 Returns:
472 List of available backends
473 """
474 lst = []
475 for backend in BUILTIN_BAKENDS:
476 try:
477 get_backend(backend)
478 except ImportError:
479 continue
480 lst.append(backend)
481 return lst