Merge HDF5 setup from part of a distributed analysis set into a joint file. Parameters ---------- joint_file : HDF5 file Joint file proc_paths : list of [str or pathlib.Path] List of files in a distributed analysis set virtual: bool, optional If True
(joint_file, proc_paths, virtual=False)
| 223 | |
| 224 | |
| 225 | def merge_setup(joint_file, proc_paths, virtual=False): |
| 226 | """ |
| 227 | Merge HDF5 setup from part of a distributed analysis set into a joint file. |
| 228 | |
| 229 | Parameters |
| 230 | ---------- |
| 231 | joint_file : HDF5 file |
| 232 | Joint file |
| 233 | proc_paths : list of [str or pathlib.Path] |
| 234 | List of files in a distributed analysis set |
| 235 | virtual: bool, optional |
| 236 | If True, merging a virtual file into a single file rather than distributed set |
| 237 | |
| 238 | """ |
| 239 | proc_path0 = pathlib.Path(proc_paths[0]) |
| 240 | logger.info("Merging setup for {}".format(joint_file)) |
| 241 | with h5py.File(str(proc_path0), mode='r') as proc_file: |
| 242 | # File metadata |
| 243 | try: |
| 244 | joint_file.attrs['set_number'] = proc_file.attrs['set_number'] |
| 245 | except KeyError: |
| 246 | joint_file.attrs['set_number'] = proc_file.attrs['file_number'] |
| 247 | joint_file.attrs['handler_name'] = proc_file.attrs['handler_name'] |
| 248 | try: |
| 249 | joint_file.attrs['writes'] = writes = proc_file.attrs['writes'] |
| 250 | except KeyError: |
| 251 | joint_file.attrs['writes'] = writes = len(proc_file['scales']['write_number']) |
| 252 | # Copy scales (distributed files all have global scales) |
| 253 | if virtual: |
| 254 | proc_file.copy('scales', joint_file) |
| 255 | else: |
| 256 | needed_hashes = [] |
| 257 | joint_scales = joint_file.create_group('scales') |
| 258 | for scalename in proc_file['scales']: |
| 259 | if 'hash_' in scalename: |
| 260 | needed_hashes.append(scalename) |
| 261 | else: |
| 262 | joint_scales.create_dataset(name=scalename, data=proc_file['scales'][scalename]) |
| 263 | # Tasks |
| 264 | joint_tasks = joint_file.create_group('tasks') |
| 265 | proc_tasks = proc_file['tasks'] |
| 266 | for taskname in proc_tasks: |
| 267 | # Setup dataset with automatic chunking |
| 268 | proc_dset = proc_tasks[taskname] |
| 269 | if virtual: |
| 270 | joint_dset = joint_tasks.create_dataset(name=proc_dset.name, data=proc_dset) |
| 271 | else: |
| 272 | spatial_shape = proc_dset.attrs['global_shape'] |
| 273 | joint_shape = (writes,) + tuple(spatial_shape) |
| 274 | joint_dset = joint_tasks.create_dataset(name=proc_dset.name, |
| 275 | shape=joint_shape, |
| 276 | dtype=proc_dset.dtype, |
| 277 | chunks=True) |
| 278 | # Dataset metadata |
| 279 | joint_dset.attrs['task_number'] = proc_dset.attrs['task_number'] |
| 280 | joint_dset.attrs['constant'] = proc_dset.attrs['constant'] |
| 281 | joint_dset.attrs['grid_space'] = proc_dset.attrs['grid_space'] |
| 282 | joint_dset.attrs['scales'] = proc_dset.attrs['scales'] |
no test coverage detected