from __future__ import annotations
from pathlib import Path
from ewokscore import Task
from ewokscore.missing_data import MissingData
from ewokstomo.tasks.nxtomo_utils import build_nxtomo_from_inputs
[docs]
class TomoBasicToNXtomo( # type: ignore[call-arg]
Task,
input_names=[
"nx_path",
"detector_data_file_paths",
"detector_data_h5_url",
"detector_data_shapes",
"detector_data_dtype",
"image_key_control",
"rotation_angle_deg",
"sample_name",
],
optional_input_names=[
"energy_kev",
"title",
"start_time",
"end_time",
"group_size",
"estimated_cor",
"detector_data_axes",
"detector_x_pixel_size_um",
"detector_y_pixel_size_um",
"sample_x_pixel_size_um",
"sample_y_pixel_size_um",
"sample_detector_distance_mm",
"source_sample_distance_mm",
"field_of_view",
"instrument_name",
"propagation_distance_mm",
"count_time_s",
"x_translation_mm",
"y_translation_mm",
"z_translation_mm",
"current_a",
"sequence_number",
],
output_names=["nx_path"],
):
"""Build a NXtomo file directly from explicit inputs (no HDF5 reads).
Units expected (raw numbers):
- energy_kev: keV
- detector_x_pixel_size_um, detector_y_pixel_size_um: micrometer
- sample_x_pixel_size_um, sample_y_pixel_size_um: micrometer
- sample_detector_distance_mm, source_sample_distance_mm, propagation_distance_mm: mm
- count_time_s: s
- rotation_angle_deg: degree
- x_translation_mm, y_translation_mm, z_translation_mm: mm
- current_a: ampere
Detector flips are described by raw Bliss `detector_data_axes`
metadata in `(ud, lr)` order, e.g. `["-z", "y"]` means no flip.
"""
[docs]
def run(self):
def _opt(name: str):
val = getattr(self.inputs, name, MissingData())
return None if isinstance(val, MissingData) else val
nx_path = Path(self.inputs.nx_path)
nx_path.parent.mkdir(parents=True, exist_ok=True)
nx_obj = build_nxtomo_from_inputs(
energy_kev=_opt("energy_kev"),
title=_opt("title"),
start_time=_opt("start_time"),
end_time=_opt("end_time"),
group_size=_opt("group_size"),
estimated_cor=_opt("estimated_cor"),
detector_data_axes=_opt("detector_data_axes"),
detector_x_pixel_size_um=_opt("detector_x_pixel_size_um"),
detector_y_pixel_size_um=_opt("detector_y_pixel_size_um"),
sample_x_pixel_size_um=_opt("sample_x_pixel_size_um"),
sample_y_pixel_size_um=_opt("sample_y_pixel_size_um"),
sample_detector_distance_mm=_opt("sample_detector_distance_mm"),
source_sample_distance_mm=_opt("source_sample_distance_mm"),
field_of_view=_opt("field_of_view"),
instrument_name=_opt("instrument_name"),
sample_name=self.inputs.sample_name,
propagation_distance_mm=_opt("propagation_distance_mm"),
detector_data_file_paths=self.inputs.detector_data_file_paths,
detector_data_h5_url=self.inputs.detector_data_h5_url,
detector_data_shapes=_opt("detector_data_shapes"),
detector_data_dtype=_opt("detector_data_dtype"),
image_key_control=self.inputs.image_key_control,
count_time_s=_opt("count_time_s"),
rotation_angle_deg=self.inputs.rotation_angle_deg,
x_translation_mm=_opt("x_translation_mm"),
y_translation_mm=_opt("y_translation_mm"),
z_translation_mm=_opt("z_translation_mm"),
current_a=_opt("current_a"),
sequence_number=_opt("sequence_number"),
)
nx_obj.save(str(nx_path), data_path="entry0000", overwrite=True)
self.outputs.nx_path = str(nx_path)
