#!/usr/bin/env python3
# -*- coding: utf-8 -*
# pylint: disable=invalid-name
"""
Created march 2026
@author: nchevaug
"""
import os
from pathlib import Path
from typing import Dict
import numpy as np
# --- AJOUT CRITIQUE POUR LA LECTURE PARAVIEW À LA VOLÉE ---
# On désactive le verrouillage de fichier de HDF5 AVANT d'importer h5py.
# Cela empêche ParaView et Python de se bloquer mutuellement.
os.environ["HDF5_USE_FILE_LOCKING"] = "FALSE"
import h5py
from pyxfem.core.config import npfloat
# Définition des types XDMF supportés et de leurs dimensions associées dans HDF5
XDMF_TYPES = {
"Scalar": 1,
"Vector": 3, # VTK/ParaView force les vecteurs en 3D
"Tensor6": 6, # Tenseur symétrique (XX, YY, ZZ, XY, YZ, XZ)
"Tensor": 9 # Tenseur complet 3x3
}
[docs]
class ParaViewXDMFWriter:
"""
Gestionnaire générique d'export HDF5 + XDMF pour ParaView.
Permet la lecture "à la volée" (On-the-fly) pendant le calcul.
"""
def __init__(self,
filename_base: str,
t2v: np.ndarray,
fields_config: Dict[str, Dict[str, str]],
overwrite: bool = False):
"""
Initialise le writer avec une configuration dynamique des champs.
"""
self.h5_filename = f"{filename_base}.h5"
self.xdmf_filename = f"{filename_base}.xdmf"
self.filename_base = os.path.basename(filename_base)
self.fields_config = fields_config
# Récupération dynamique de la précision des entiers pour le maillage (évite les bugs ParaView)
self.int_precision = t2v.dtype.itemsize
# Validation rapide de la configuration
for name, config in self.fields_config.items():
if config["type"] not in XDMF_TYPES:
raise ValueError(f"Type XDMF non supporté pour {name}: {config['type']}")
if config["center"] not in ["Node", "Cell"]:
raise ValueError(f"Center XDMF non supporté pour {name}: {config['center']}")
if not overwrite and (os.path.exists(self.h5_filename) or os.path.exists(self.xdmf_filename)):
raise FileExistsError(
f"Warning! The files {filename_base}.* already exist. "
f"Change the output name or pass overwrite=True to force overwrite."
)
self.ncells = t2v.shape[0]
self.steps = [] # Stockera (step, time, nverts)
# On initialise le fichier HDF5 et on y écrit la connectivité.
# Le bloc 'with' garantit la fermeture immédiate du fichier (Flush sécurisé).
with h5py.File(self.h5_filename, 'w') as h5f:
h5f.create_dataset("Mesh/Connectivity", data=t2v, compression="gzip")
[docs]
def write_step(self, step: int, time: float, xy: np.ndarray, **field_data: np.ndarray):
"""
Écrit un pas de temps de manière sécurisée et met à jour le fichier XDMF.
"""
nverts = xy.shape[0]
self.steps.append((step, time, nverts)) # On sauvegarde nverts au cas où le maillage évolue
# Vérification stricte du contrat
missing_fields = set(self.fields_config.keys()) - set(field_data.keys())
if missing_fields:
raise KeyError(f"Données manquantes pour l'export: {missing_fields}")
extra_fields = set(field_data.keys()) - set(self.fields_config.keys())
if extra_fields:
raise KeyError(f"Unknown Key for export: {extra_fields}")
# Ouverture en mode 'a' (append). Les données sont "flushées" sur le disque à la sortie du 'with'
with h5py.File(self.h5_filename, 'a') as h5f:
grp = h5f.create_group(f"States/Step_{step}")
# --- GÉOMÉTRIE (Padding 2D -> 3D) ---
xy_3d = np.zeros((nverts, 3), dtype=xy.dtype)
xy_3d[:, :2] = xy
grp.create_dataset("Coordinates", data=xy_3d, compression="gzip")
# --- CHAMPS GÉNÉRIQUES ---
for name, data in field_data.items():
field_type = self.fields_config[name]["type"]
expected_dim = XDMF_TYPES[field_type]
n_items = nverts if self.fields_config[name]["center"] == "Node" else self.ncells
# Formatage selon le type demandé par ParaView
if field_type == "Scalar":
out_data = data.reshape(n_items, expected_dim)
elif field_type == "Vector":
if data.shape[-1] == 2:
out_data = np.zeros((n_items, expected_dim), dtype=data.dtype)
out_data[:, :2] = data.reshape(n_items, 2)
else:
out_data = data.reshape(n_items, expected_dim)
elif field_type == "Tensor":
if data.shape[-2:] == (2, 2):
out_data = np.zeros((n_items, expected_dim), dtype=data.dtype)
out_data[:, 0] = data[..., 0, 0] # XX
out_data[:, 1] = data[..., 0, 1] # XY
out_data[:, 3] = data[..., 1, 0] # YX
out_data[:, 4] = data[..., 1, 1] # YY
else:
out_data = data.reshape(n_items, expected_dim)
elif field_type == "Tensor6":
if data.shape[-2:] == (2, 2):
out_data = np.zeros((n_items, expected_dim), dtype=data.dtype)
out_data[:, 0] = data[..., 0, 0] # XX
out_data[:, 1] = data[..., 1, 1] # YY
out_data[:, 3] = data[..., 0, 1] # XY
else:
out_data = data.reshape(n_items, expected_dim)
# Sécurité finale sur les dimensions
assert out_data.shape == (n_items, expected_dim), \
f"Erreur de dimension pour {name}: attendu {(n_items, expected_dim)}, obtenu {out_data.shape}"
grp.create_dataset(name, data=out_data, compression="gzip")
# --- RÉÉCRITURE DU XDMF À LA VOLÉE ---
# Le HDF5 est fermé et sécurisé. On met à jour le fichier texte XML pour ParaView.
self._write_master_xdmf()
def _write_master_xdmf(self):
"""Méthode interne pour régénérer le fichier XDMF avec tous les pas de temps actuels."""
float_precision = np.dtype(npfloat).itemsize
h5_file = f"{self.filename_base}.h5"
xdmf_content = [
'<?xml version="1.0" ?>',
'<!DOCTYPE Xdmf SYSTEM "Xdmf.dtd" []>',
'<Xdmf Version="2.0">',
' <Domain>',
' <Grid Name="TimeSeries" GridType="Collection" CollectionType="Temporal">'
]
for step, time, nverts in self.steps:
grid = [
f' <Grid Name="Step_{step}" GridType="Uniform">',
f' <Time Value="{time:e}" />',
f' <Topology TopologyType="Triangle" NumberOfElements="{self.ncells}">',
f' <DataItem Dimensions="{self.ncells} 3" NumberType="Int" Precision="{self.int_precision}" Format="HDF">',
f' {h5_file}:/Mesh/Connectivity',
f' </DataItem>',
f' </Topology>',
f' <Geometry GeometryType="XYZ">',
f' <DataItem Dimensions="{nverts} 3" NumberType="Float" Precision="{float_precision}" Format="HDF">',
f' {h5_file}:/States/Step_{step}/Coordinates',
f' </DataItem>',
f' </Geometry>'
]
# Ajout dynamique des attributs
for name, config in self.fields_config.items():
center = config["center"]
field_type = config["type"]
dim = XDMF_TYPES[field_type]
n_items = nverts if center == "Node" else self.ncells
grid.extend([
f' <Attribute Name="{name}" AttributeType="{field_type}" Center="{center}">',
f' <DataItem Dimensions="{n_items} {dim}" NumberType="Float" Precision="{float_precision}" Format="HDF">',
f' {h5_file}:/States/Step_{step}/{name}',
f' </DataItem>',
f' </Attribute>'
])
grid.append(' </Grid>')
xdmf_content.extend(grid)
xdmf_content.extend([
' </Grid>',
' </Domain>',
'</Xdmf>'
])
# Écriture atomique : on écrase l'ancien fichier texte par le nouveau complet
Path(self.xdmf_filename).write_text('\n'.join(xdmf_content), encoding='utf-8')
[docs]
def finalize(self):
"""
Génère le fichier XML (XDMF) final.
Conservé uniquement pour la rétrocompatibilité de l'interface.
"""
self._write_master_xdmf()