import shutil
from pathlib import Path
import h5py
import numpy as np
import pytest
from ewoksutils.exceptions import TaskExecutionError
from ..tasks.read_fscan2d import ReadFscan2d
from .conftest import write_integration_output as _write_integration_output
BLISS_MASTER = Path(__file__).parent / "data" / "test_fscan2d.h5"
# BLISS_MASTER records a 5x5 fscan2d with "hry" as the fast motor (5 points
# from 0 in steps of 0.01) and "hrz" as the slow one (5 points from 0.93 in
# steps of 0.01). Both are translations, but the task only matches the given
# motor names against the recorded fast/slow ones, so either can play the
# rotation role.
N_POINTS = 25
# Parameters to use to generate test ai file
N_RADIAL = 1500
RADIAL_AXIS = np.linspace(0.003, 5.477, N_RADIAL, dtype=np.float32)
INTENSITY = np.arange(N_POINTS * N_RADIAL, dtype=np.float32).reshape(N_POINTS, N_RADIAL)
[docs]
@pytest.fixture(scope="module")
def ai_file(tmp_path_factory):
"""An azimuthal-integration output with one frame per scan point, built
once and shared across tests since they only ever read it."""
path = tmp_path_factory.mktemp("ai") / "ai.h5"
_write_integration_output(path, radial_axis=RADIAL_AXIS, intensity=INTENSITY)
return path
def _recorded_positions(motor_name):
"""The positions recorded for `motor_name` in BLISS_MASTER."""
with h5py.File(BLISS_MASTER, "r") as f:
return f[f"1.1/measurement/{motor_name}"][()]
def _copy_master_with_overrides(tmp_path, overrides):
"""Copy BLISS_MASTER, replacing or adding the datasets named by the HDF5
paths in `overrides`, to build a master that differs from the recorded
scan."""
master = tmp_path / "master.h5"
shutil.copy(BLISS_MASTER, master)
with h5py.File(master, "r+") as f:
for path, value in overrides.items():
if path in f:
del f[path]
f[path] = value
return master
def _make_task(ai_file, master=BLISS_MASTER, **input_overrides):
inputs = {
"bliss_master_path": str(master),
"integration_output_path": str(ai_file),
"scan_number": 1,
"translation_motor_name": "hrz",
"rotation_motor_name": "hry",
"integration_intensity_path": "/intensities",
"integration_radial_axis_path": "/2th_deg",
}
inputs.update(input_overrides)
return ReadFscan2d(inputs=inputs)
[docs]
def test_run_reads_raw_data_rotation_is_fast_motor(ai_file):
"""The caller declares the recorded fast motor as the rotation axis: raw
arrays pass through unreshaped, and grid info uses the bin-centered
formula for rotation, the plain inclusive-endpoint formula for
translation."""
task = _make_task(ai_file)
task.execute()
np.testing.assert_array_equal(
task.outputs.rotation_angles, _recorded_positions("hry")
)
np.testing.assert_array_equal(
task.outputs.translation_values, _recorded_positions("hrz")
)
assert task.outputs.rotation_angles.shape == (N_POINTS,)
np.testing.assert_array_equal(task.outputs.integration_intensity_values, INTENSITY)
assert task.outputs.integration_radial_axis == "2th_deg"
np.testing.assert_array_equal(
task.outputs.integration_radial_axis_values, RADIAL_AXIS
)
# Grid values derived from the fscan_parameters of BLISS_MASTER.
assert task.outputs.rotation_grid_params == pytest.approx((0.005, 0.045, 5))
assert task.outputs.translation_grid_params == pytest.approx((0.93, 0.97, 5))
[docs]
def test_run_reads_raw_data_rotation_is_slow_motor(ai_file):
"""The caller declares the recorded fast motor as the translation axis
instead, which is the physical case for this scan: the grid info
assignment swaps accordingly, still with no reshaping of the raw
arrays."""
task = _make_task(
ai_file,
translation_motor_name="hry",
rotation_motor_name="hrz",
)
task.execute()
np.testing.assert_array_equal(
task.outputs.rotation_angles, _recorded_positions("hrz")
)
np.testing.assert_array_equal(
task.outputs.translation_values, _recorded_positions("hry")
)
# Grid values derived from the fscan_parameters of BLISS_MASTER.
assert task.outputs.rotation_grid_params == pytest.approx((0.93, 0.97, 5))
assert task.outputs.translation_grid_params == pytest.approx((0.005, 0.045, 5))
[docs]
def test_run_motor_mismatch_raises(ai_file):
"""Either rotation_motor_name or translation_motor_name does not match the
fast/slow motors recorded for the scan raises a ValueError."""
# "backscat" is recorded for the scan, so it is read successfully and only
# then fails to match either the fast or the slow motor.
task = _make_task(ai_file, translation_motor_name="backscat")
with pytest.raises(TaskExecutionError, match="does not match") as excinfo:
task.execute()
assert isinstance(excinfo.value.__cause__, ValueError)
[docs]
def test_run_frame_count_mismatch_raises(tmp_path):
"""An integration output covering only part of the scan raises ValueError."""
partial_ai_file = tmp_path / "partial_ai.h5"
_write_integration_output(
partial_ai_file,
radial_axis=RADIAL_AXIS,
intensity=INTENSITY[: N_POINTS // 5],
)
task = _make_task(partial_ai_file)
with pytest.raises(TaskExecutionError, match="integrated frames") as excinfo:
task.execute()
assert isinstance(excinfo.value.__cause__, ValueError)
[docs]
def test_run_point_count_mismatch_raises(tmp_path, ai_file):
"""The recorded array length not matching fast_npoints * slow_npoints
raises a ValueError."""
# fast_npoints=4 gives 4*5=20 expected points against the 25 recorded.
master = _copy_master_with_overrides(
tmp_path, {"1.1/instrument/fscan_parameters/fast_npoints": 4}
)
task = _make_task(ai_file, master=master)
with pytest.raises(
TaskExecutionError, match="Expected 20 translation positions"
) as excinfo:
task.execute()
assert isinstance(excinfo.value.__cause__, ValueError)
[docs]
def test_run_wrong_integration_path_raises(ai_file):
"""A radial axis or intensity HDF5 path that does not exist raises ValueError."""
task = _make_task(ai_file, integration_intensity_path="/does/not/exist")
with pytest.raises(TaskExecutionError, match="/does/not/exist") as excinfo:
task.execute()
assert isinstance(excinfo.value.__cause__, ValueError)