Coverage for pyxfem/postpro/ioh5.py: 15%

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -* 

3# pylint: disable=invalid-name 

4""" 

5Created march 2026 

6@author: nchevaug 

7""" 

8import os 

9from pathlib import Path 

10from typing import Dict 

11import numpy as np 

12 

13 

14# --- AJOUT CRITIQUE POUR LA LECTURE PARAVIEW À LA VOLÉE --- 

15# On désactive le verrouillage de fichier de HDF5 AVANT d'importer h5py. 

16# Cela empêche ParaView et Python de se bloquer mutuellement. 

17os.environ["HDF5_USE_FILE_LOCKING"] = "FALSE" 

18 

19import h5py 

20from pyxfem.core.config import npfloat 

21 

22 

23# Définition des types XDMF supportés et de leurs dimensions associées dans HDF5 

24XDMF_TYPES = { 

25 "Scalar": 1, 

26 "Vector": 3, # VTK/ParaView force les vecteurs en 3D 

27 "Tensor6": 6, # Tenseur symétrique (XX, YY, ZZ, XY, YZ, XZ) 

28 "Tensor": 9 # Tenseur complet 3x3 

29} 

30 

31class ParaViewXDMFWriter: 

32 """ 

33 Gestionnaire générique d'export HDF5 + XDMF pour ParaView. 

34 Permet la lecture "à la volée" (On-the-fly) pendant le calcul. 

35 """ 

36 def __init__(self, 

37 filename_base: str, 

38 t2v: np.ndarray, 

39 fields_config: Dict[str, Dict[str, str]], 

40 overwrite: bool = False): 

41 """ 

42 Initialise le writer avec une configuration dynamique des champs. 

43 """ 

44 self.h5_filename = f"{filename_base}.h5" 

45 self.xdmf_filename = f"{filename_base}.xdmf" 

46 self.filename_base = os.path.basename(filename_base) 

47 self.fields_config = fields_config 

48 

49 # Récupération dynamique de la précision des entiers pour le maillage (évite les bugs ParaView) 

50 self.int_precision = t2v.dtype.itemsize 

51 

52 # Validation rapide de la configuration 

53 for name, config in self.fields_config.items(): 

54 if config["type"] not in XDMF_TYPES: 

55 raise ValueError(f"Type XDMF non supporté pour {name}: {config['type']}") 

56 if config["center"] not in ["Node", "Cell"]: 

57 raise ValueError(f"Center XDMF non supporté pour {name}: {config['center']}") 

58 

59 if not overwrite and (os.path.exists(self.h5_filename) or os.path.exists(self.xdmf_filename)): 

60 raise FileExistsError( 

61 f"Warning! The files {filename_base}.* already exist. " 

62 f"Change the output name or pass overwrite=True to force overwrite." 

63 ) 

64 

65 self.ncells = t2v.shape[0] 

66 self.steps = [] # Stockera (step, time, nverts) 

67 

68 # On initialise le fichier HDF5 et on y écrit la connectivité. 

69 # Le bloc 'with' garantit la fermeture immédiate du fichier (Flush sécurisé). 

70 with h5py.File(self.h5_filename, 'w') as h5f: 

71 h5f.create_dataset("Mesh/Connectivity", data=t2v, compression="gzip") 

72 

73 def write_step(self, step: int, time: float, xy: np.ndarray, **field_data: np.ndarray): 

74 """ 

75 Écrit un pas de temps de manière sécurisée et met à jour le fichier XDMF. 

76 """ 

77 nverts = xy.shape[0] 

78 self.steps.append((step, time, nverts)) # On sauvegarde nverts au cas où le maillage évolue 

79 

80 # Vérification stricte du contrat 

81 missing_fields = set(self.fields_config.keys()) - set(field_data.keys()) 

82 if missing_fields: 

83 raise KeyError(f"Données manquantes pour l'export: {missing_fields}") 

84 extra_fields = set(field_data.keys()) - set(self.fields_config.keys()) 

85 if extra_fields: 

86 raise KeyError(f"Unknown Key for export: {extra_fields}") 

87 

88 # Ouverture en mode 'a' (append). Les données sont "flushées" sur le disque à la sortie du 'with' 

89 with h5py.File(self.h5_filename, 'a') as h5f: 

90 grp = h5f.create_group(f"States/Step_{step}") 

91 

92 # --- GÉOMÉTRIE (Padding 2D -> 3D) --- 

93 xy_3d = np.zeros((nverts, 3), dtype=xy.dtype) 

94 xy_3d[:, :2] = xy 

95 grp.create_dataset("Coordinates", data=xy_3d, compression="gzip") 

96 

97 # --- CHAMPS GÉNÉRIQUES --- 

98 for name, data in field_data.items(): 

99 field_type = self.fields_config[name]["type"] 

100 expected_dim = XDMF_TYPES[field_type] 

101 n_items = nverts if self.fields_config[name]["center"] == "Node" else self.ncells 

102 

103 # Formatage selon le type demandé par ParaView 

104 if field_type == "Scalar": 

105 out_data = data.reshape(n_items, expected_dim) 

106 

107 elif field_type == "Vector": 

108 if data.shape[-1] == 2: 

109 out_data = np.zeros((n_items, expected_dim), dtype=data.dtype) 

110 out_data[:, :2] = data.reshape(n_items, 2) 

111 else: 

112 out_data = data.reshape(n_items, expected_dim) 

113 

114 elif field_type == "Tensor": 

115 if data.shape[-2:] == (2, 2): 

116 out_data = np.zeros((n_items, expected_dim), dtype=data.dtype) 

117 out_data[:, 0] = data[..., 0, 0] # XX 

118 out_data[:, 1] = data[..., 0, 1] # XY 

119 out_data[:, 3] = data[..., 1, 0] # YX 

120 out_data[:, 4] = data[..., 1, 1] # YY 

121 else: 

122 out_data = data.reshape(n_items, expected_dim) 

123 

124 elif field_type == "Tensor6": 

125 if data.shape[-2:] == (2, 2): 

126 out_data = np.zeros((n_items, expected_dim), dtype=data.dtype) 

127 out_data[:, 0] = data[..., 0, 0] # XX 

128 out_data[:, 1] = data[..., 1, 1] # YY 

129 out_data[:, 3] = data[..., 0, 1] # XY 

130 else: 

131 out_data = data.reshape(n_items, expected_dim) 

132 

133 # Sécurité finale sur les dimensions 

134 assert out_data.shape == (n_items, expected_dim), \ 

135 f"Erreur de dimension pour {name}: attendu {(n_items, expected_dim)}, obtenu {out_data.shape}" 

136 

137 grp.create_dataset(name, data=out_data, compression="gzip") 

138 

139 # --- RÉÉCRITURE DU XDMF À LA VOLÉE --- 

140 # Le HDF5 est fermé et sécurisé. On met à jour le fichier texte XML pour ParaView. 

141 self._write_master_xdmf() 

142 

143 def _write_master_xdmf(self): 

144 """Méthode interne pour régénérer le fichier XDMF avec tous les pas de temps actuels.""" 

145 float_precision = np.dtype(npfloat).itemsize 

146 h5_file = f"{self.filename_base}.h5" 

147 

148 xdmf_content = [ 

149 '<?xml version="1.0" ?>', 

150 '<!DOCTYPE Xdmf SYSTEM "Xdmf.dtd" []>', 

151 '<Xdmf Version="2.0">', 

152 ' <Domain>', 

153 ' <Grid Name="TimeSeries" GridType="Collection" CollectionType="Temporal">' 

154 ] 

155 

156 for step, time, nverts in self.steps: 

157 grid = [ 

158 f' <Grid Name="Step_{step}" GridType="Uniform">', 

159 f' <Time Value="{time:e}" />', 

160 f' <Topology TopologyType="Triangle" NumberOfElements="{self.ncells}">', 

161 f' <DataItem Dimensions="{self.ncells} 3" NumberType="Int" Precision="{self.int_precision}" Format="HDF">', 

162 f' {h5_file}:/Mesh/Connectivity', 

163 f' </DataItem>', 

164 f' </Topology>', 

165 f' <Geometry GeometryType="XYZ">', 

166 f' <DataItem Dimensions="{nverts} 3" NumberType="Float" Precision="{float_precision}" Format="HDF">', 

167 f' {h5_file}:/States/Step_{step}/Coordinates', 

168 f' </DataItem>', 

169 f' </Geometry>' 

170 ] 

171 

172 # Ajout dynamique des attributs 

173 for name, config in self.fields_config.items(): 

174 center = config["center"] 

175 field_type = config["type"] 

176 dim = XDMF_TYPES[field_type] 

177 n_items = nverts if center == "Node" else self.ncells 

178 

179 grid.extend([ 

180 f' <Attribute Name="{name}" AttributeType="{field_type}" Center="{center}">', 

181 f' <DataItem Dimensions="{n_items} {dim}" NumberType="Float" Precision="{float_precision}" Format="HDF">', 

182 f' {h5_file}:/States/Step_{step}/{name}', 

183 f' </DataItem>', 

184 f' </Attribute>' 

185 ]) 

186 

187 grid.append(' </Grid>') 

188 xdmf_content.extend(grid) 

189 

190 xdmf_content.extend([ 

191 ' </Grid>', 

192 ' </Domain>', 

193 '</Xdmf>' 

194 ]) 

195 

196 # Écriture atomique : on écrase l'ancien fichier texte par le nouveau complet 

197 Path(self.xdmf_filename).write_text('\n'.join(xdmf_content), encoding='utf-8') 

198 

199 def finalize(self): 

200 """ 

201 Génère le fichier XML (XDMF) final. 

202 Conservé uniquement pour la rétrocompatibilité de l'interface. 

203 """ 

204 self._write_master_xdmf()