Prepare data represented as numpy array, dask array (no change) or HDF5 dataset
(data_dask, data_representation, tmpdir, *, chunked_HDF5=True)
| 302 | |
| 303 | |
| 304 | def _create_xrf_data(data_dask, data_representation, tmpdir, *, chunked_HDF5=True): |
| 305 | """Prepare data represented as numpy array, dask array (no change) or HDF5 dataset""" |
| 306 | if data_representation == "numpy_array": |
| 307 | data = data_dask.compute() |
| 308 | elif data_representation == "dask_array": |
| 309 | data = data_dask |
| 310 | elif data_representation == "hdf5_file_dset": |
| 311 | os.chdir(tmpdir) |
| 312 | fln = f"test-{uuid.uuid4()}.h5" # Include UUID in the file name |
| 313 | dset_name = "level1/level2" |
| 314 | with h5py.File(fln, "w") as f: |
| 315 | # In this test all computations are performed using 'float64' precision, |
| 316 | # so we create the dataset with dtype="float64" for consistency. |
| 317 | kwargs = {"shape": data_dask.shape, "dtype": "float64"} |
| 318 | if chunked_HDF5: |
| 319 | kwargs.update({"chunks": data_dask.chunksize}) |
| 320 | dset = f.create_dataset(dset_name, **kwargs) |
| 321 | dset[:, :, :] = data_dask.compute() |
| 322 | data = RawHDF5Dataset(fln, dset_name, shape=data_dask.shape) |
| 323 | else: |
| 324 | raise RuntimeError( |
| 325 | f"Error in test parameter: unknown value of 'data_representation' = {data_representation}" |
| 326 | ) |
| 327 | |
| 328 | return data |
| 329 | |
| 330 | |
| 331 | def _create_xrf_mask(data_shape, apply_mask, select_area): |
no test coverage detected