import configparser
import os
import shutil
import h5py
from typing import Tuple, Optional
from coseda.nexus.images import ensure_image_nxdata
from coseda.nexus.process import write_beam_incident_energy, write_nxprocess_import
from coseda.nexus.indices import ensure_image_key
from coseda.nexus.groups import ensure_nexus_parents
from coseda.nexus.detector import write_detector_geometry
from coseda.nexus.logs import ensure_dense_logs
from coseda.nexus.goniometer import ensure_goniometer_transforms, get_goniometer_transform_order
[docs]
def move_plain_h5(ini_path: str) -> str:
"""
Move and rename the original HDF5 file according to the INI's Paths section.
Reads from the INI [Paths]:
- 'hdf5_file' / 'hdf5_input' / 'hdf5_file_path': original HDF5 path.
- 'output_folder' / 'output_directory': target directory.
Renames the HDF5 to match the INI basename: <ini_basename>.h5,
moves it into the output folder (creating it if needed),
and returns the new HDF5 file path.
"""
config = configparser.ConfigParser()
config.read(ini_path)
if "Paths" not in config:
raise KeyError(f"No [Paths] section found in INI: {ini_path}")
paths = config["Paths"]
# Determine original HDF5 path
original = (
paths.get("originalfile_path")
)
if not original:
raise KeyError(
"HDF5 file path not found in INI [Paths] (checked 'originalfile')."
)
if not os.path.isfile(original):
raise FileNotFoundError(f"Original HDF5 not found: {original}")
# Determine paths from INI
outputfolder = paths.get("outputfolder")
h5file_name = paths.get("h5file")
configfile_ini = paths.get("configfile_path") or ini_path
parent_dir = os.path.dirname(os.path.abspath(configfile_ini))
# Create output folder under parent directory
if outputfolder:
output_dir = os.path.join(parent_dir, outputfolder)
else:
output_dir = parent_dir
os.makedirs(output_dir, exist_ok=True)
# Determine new HDF5 file name and path
if not h5file_name:
h5file_name = os.path.splitext(os.path.basename(ini_path))[0] + ".h5"
new_path = os.path.join(output_dir, h5file_name)
# Move and rename
shutil.move(original, new_path)
# Update INI with new HDF5 path
# Determine which key was used for the original path
for key in ("hdf5_file", "hdf5_input", "hdf5_file_path"):
if paths.get(key):
config.set("Paths", key, new_path)
break
# Write updates back to the INI file
with open(ini_path, "w") as configfile:
config.write(configfile)
# Add NeXus image view and beam energy if possible.
try:
goniometer_transform_order = get_goniometer_transform_order(config)
with h5py.File(new_path, "r+") as h5file:
ensure_nexus_parents(h5file)
ensure_image_nxdata(h5file)
ensure_image_key(h5file)
ensure_dense_logs(h5file)
ensure_goniometer_transforms(h5file, goniometer_transform_order)
write_nxprocess_import(
h5file,
program="coseda.importers.import_h5",
input_path=original,
output_path=new_path,
)
try:
av = float(config.get("AcquisitionDetails", "acceleration_voltage"))
except Exception:
av = None
if av is not None:
write_beam_incident_energy(h5file, av)
try:
camera_length = float(config.get("AcquisitionDetails", "camera_length"))
except Exception:
camera_length = None
try:
camera_length_correction = float(config.get("AcquisitionDetails", "camera_length_correction"))
except Exception:
camera_length_correction = 1.0
if camera_length is not None:
write_detector_geometry(h5file, camera_length, camera_length_correction)
except Exception:
pass
return new_path
[docs]
def check_image_dataset(
h5_path: str,
dataset_path: str = "/entry/data/images"
) -> Tuple[bool, Optional[Tuple[int, int, int]]]:
"""
Check whether `dataset_path` exists in the HDF5 file at `h5_path`,
and whether it has exactly 3 dimensions (frames × height × width).
Returns:
(True, shape) if it exists and ndim == 3
(False, None) if it doesn’t exist or file cannot be opened
(False, shape) if it exists but shape is not 3D
"""
try:
with h5py.File(h5_path, "r") as f:
if dataset_path not in f:
return False, None
dset = f[dataset_path]
shape = dset.shape
# Expecting (n_frames, height, width)
if len(shape) != 3:
return False, shape
return True, shape
except (OSError, IOError):
return False, None
[docs]
def is_dataset_chunked_first_dim(
h5_path: str,
dataset_path: str = "/entry/data/images"
) -> Tuple[bool, Optional[Tuple[int, ...]]]:
"""
Check whether `dataset_path` in the HDF5 file at `h5_path` is stored with
chunking (i.e., has a non-None chunks tuple). Specifically tests that the
dataset uses HDF5 chunked storage, which implies chunking along all dims,
including the first.
Returns:
(True, chunks) if the dataset exists and is chunked (chunks is the chunk shape tuple)
(False, None) if the dataset does not exist, cannot be opened, or is not chunked
"""
try:
with h5py.File(h5_path, "r") as f:
if dataset_path not in f:
return False, None
dset = f[dataset_path]
chunks = dset.chunks
if chunks is not None:
return True, chunks
return False, None
except (OSError, IOError):
return False, None
[docs]
def get_dataset_chunk_info(
h5_path: str,
dataset_path: str = "/entry/data/images"
) -> Tuple[bool, Optional[Tuple[int, ...]]]:
"""
Check if `dataset_path` exists, is 3D, and is chunked in the HDF5 file at `h5_path`.
Returns:
(True, chunks) if the dataset exists, is 3D, and uses chunked storage (chunks is tuple of chunk sizes)
(True, ()) if it exists, is 3D, but uses contiguous storage (no chunking)
(False, None) if the dataset is missing, not 3D, or the file cannot be opened
"""
try:
with h5py.File(h5_path, "r") as f:
if dataset_path not in f:
return False, None
dset = f[dataset_path]
shape = dset.shape
# Ensure it's 3D
if len(shape) != 3:
return False, None
chunks = dset.chunks or ()
if chunks:
return True, chunks
return True, ()
except (OSError, IOError):
return False, None