import shutil
from pathlib import Path
import pytest
from ewokstomo.tasks.reconstruct_volume import ReconstructVolume
from ewoks import execute_graph
from importlib.resources import files
DATA_ROOT = Path(__file__).resolve().parent / "data"
RAW_COLLECTION = "TestEwoksTomo"
ROTATION_AXIS_POSITION = 8.0
[docs]
def get_json_file(file_name: str) -> Path:
return Path(str(files("ewokstomo.workflows").joinpath(file_name)))
[docs]
def get_data_file(file_name):
return str(DATA_ROOT / "RAW_DATA" / RAW_COLLECTION / file_name / f"{file_name}.h5")
[docs]
def get_data_dir(scan_name: str) -> Path:
return DATA_ROOT / "PROCESSED_DATA" / RAW_COLLECTION / scan_name
[docs]
@pytest.fixture
def tmp_dataset_path(tmp_path) -> Path:
src_dir = get_data_dir("TestEwoksTomo_0010")
dst_dir = (
tmp_path
/ "visitor"
/ "ma1234"
/ "id00"
/ "20251201"
/ "PROCESSED_DATA"
/ "TestEwoksTomo"
/ "TestEwoksTomo_0010"
)
shutil.copytree(src_dir, dst_dir)
proj_dir = dst_dir / "projections"
proj_dir.mkdir(exist_ok=True)
volume_dir = dst_dir / "volumes"
if volume_dir.exists():
shutil.rmtree(volume_dir)
# remove any existing darks/flats and gallery
for pattern in ("*_darks.hdf5", "*_flats.hdf5", "gallery"):
for f in proj_dir.glob(pattern):
if f.is_dir():
shutil.rmtree(f)
else:
f.unlink()
references_dir = dst_dir / "references"
if references_dir.exists():
shutil.rmtree(references_dir)
return dst_dir
[docs]
@pytest.mark.parametrize("Task", [ReconstructVolume])
def test_reconstructvolume_task_outputs(Task, tmp_dataset_path):
nx = tmp_dataset_path / "projections" / "TestEwoksTomo_0010.nx"
nx_path = "dontexist.nx"
nabu_conf = {
"dataset": {"location": None},
"reconstruction": {
"start_z": "middle",
"end_z": "middle",
"rotation_axis_position": ROTATION_AXIS_POSITION,
},
"output": {"location": ""},
}
with pytest.raises(FileNotFoundError):
Task(
inputs={
"config_dict": nabu_conf,
"slice_index_range": None,
"nx_path": nx_path,
}
).run()
nx_path = str(nx)
task = Task(
inputs={"config_dict": nabu_conf, "slice_index_range": None, "nx_path": nx_path}
)
task.execute()
rec_path = Path(task.outputs.reconstructed_volume_path)
assert rec_path.exists(), "Reconstructed volume path does not exist"
assert rec_path.is_file(), "Reconstructed volume path is not a file"
assert rec_path.suffix == ".hdf5"
volumes_dir = rec_path.parent
stem = Path(nx_path).stem
# Directory name: sample_XXXX_absorption_32Bit_hdf5
dirname = volumes_dir.name
assert dirname.startswith(stem)
assert "_absorption_" in dirname
assert dirname.endswith("hdf5")
main_file = volumes_dir / f"{stem}.hdf5"
assert main_file.exists(), "Main reconstructed HDF5 file missing"
subdir = volumes_dir / stem
assert subdir.exists(), "Subdirectory with slice file missing"
assert subdir.is_dir()
slice_files = list(subdir.glob(f"{stem}_*.hdf5"))
assert len(slice_files) == 1, "Expected exactly one slice HDF5 file"
slice_file = slice_files[0]
assert slice_file.stem.startswith(f"{stem}_")
nabu_dict = task.outputs.nabu_dict
assert isinstance(nabu_dict, dict)
assert nabu_dict["dataset"]["location"] == str(nx_path)
assert (
nabu_dict["reconstruction"]["rotation_axis_position"] == ROTATION_AXIS_POSITION
)
assert str(tmp_dataset_path / "volumes") in nabu_dict["output"]["location"]
assert nabu_dict["output"]["file_prefix"] == rec_path.stem
[docs]
@pytest.mark.parametrize("workflow", ["reconstruction_volume.json"])
def test_reconstructvolume_and_convert_workflow_outputs(workflow, tmp_dataset_path):
h5_file_path = get_data_file("TestEwoksTomo_0010")
nx_path = tmp_dataset_path / "projections" / "TestEwoksTomo_0010.nx"
nx = str(nx_path)
workflow_file_path = get_json_file(workflow)
nabu_conf = {
"dataset": {"location": None},
"reconstruction": {
"rotation_axis_position": ROTATION_AXIS_POSITION,
},
"output": {"location": "", "file_format": "hdf5"},
}
reconstructed = execute_graph(
workflow_file_path,
inputs=[
{
"task_identifier": "ewokstomo.tasks.nxtomomill.H5ToNx",
"name": "bliss_hdf5_path",
"value": h5_file_path,
},
{
"task_identifier": "ewokstomo.tasks.reconstruct_volume.ReconstructVolume",
"name": "config_dict",
"value": nabu_conf,
},
{
"task_identifier": "ewokstomo.tasks.reconstruct_volume.ReconstructVolume",
"name": "slice_index_range",
"value": None,
},
{
"task_identifier": "ewokstomo.tasks.nxtomomill.H5ToNx",
"name": "nx_path",
"value": nx,
},
{
"task_identifier": "ewokstomo.tasks.reconstruct_volume.ReconstructVolume",
"name": "nx_path",
"value": nx,
},
],
outputs=[{"all": True}],
)
rec_path = Path(reconstructed["converted_volume_path"])
assert rec_path.exists(), "Reconstructed volume path does not exist"
assert rec_path.is_file(), "Reconstructed volume path is not a file"
assert rec_path.suffix == ".tiff"
volumes_dir = rec_path.parent
stem = Path(nx_path).stem
# Directory name: sample_XXXX_absorption_16Bit_tif
dirname = volumes_dir.name
assert dirname.startswith(stem)
assert "_absorption_" in dirname
assert dirname.endswith("tiff")
main_file = volumes_dir / f"{stem}.tiff"
assert main_file.exists(), "Main reconstructed TIF file missing"
nabu_dict = reconstructed["nabu_dict"]
assert isinstance(nabu_dict, dict)
assert nabu_dict["dataset"]["location"] == str(nx_path)
assert (
nabu_dict["reconstruction"]["rotation_axis_position"] == ROTATION_AXIS_POSITION
)
assert str(tmp_dataset_path / "volumes") in nabu_dict["output"]["location"]
assert nabu_dict["output"]["file_prefix"] == rec_path.stem
