945 lines
35 KiB
Python
945 lines
35 KiB
Python
"""Unit tests for the Timepix device with a mocked backend."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
import time
|
|
from contextlib import ExitStack
|
|
from unittest import mock
|
|
|
|
import numpy as np
|
|
import ophyd
|
|
import pytest
|
|
from bec_lib.devicemanager import ScanInfo
|
|
from bec_lib.messages import ScanStatusMessage
|
|
from bec_server.device_server.tests.utils import DMMock
|
|
from bec_server.scan_server.scans.scan_base import ScanInfo as ScanServerScanInfo
|
|
from ophyd import DeviceStatus, Staged, StatusBase
|
|
from ophyd_devices.interfaces.base_classes.psi_device_base import DeviceStoppedError
|
|
from ophyd_devices.tests.utils import MockPV, patch_dual_pvs
|
|
|
|
from superxas_bec.devices.timepix.timepix import (
|
|
ACQUIRESTATUS,
|
|
DATASOURCE,
|
|
EXPOSUREMODE,
|
|
FILEWRITEMODE,
|
|
TRIGGERMODE,
|
|
TRIGGERSOURCE,
|
|
TDCEdge,
|
|
TDCOuput,
|
|
Timepix,
|
|
window_mean,
|
|
)
|
|
from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_client import TimePixFlyStatus
|
|
from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_interface import (
|
|
NetAddresses,
|
|
PixelMap,
|
|
)
|
|
from superxas_bec.devices.utils.utils import fetch_scan_info
|
|
|
|
# pylint: disable=protected-access
|
|
# pylint: disable=redefined-outer-name
|
|
|
|
|
|
class FakeBackendClient:
|
|
"""Backend client double that can finish status callbacks on demand."""
|
|
|
|
def __init__(self):
|
|
self.status = TimePixFlyStatus.CONFIG
|
|
self._status_callbacks = {}
|
|
|
|
def add_status_callback(self, status, success, error, run=True):
|
|
"""Register status callback with optional immediate completion."""
|
|
if run and self.status in success:
|
|
status.set_finished()
|
|
return
|
|
self._status_callbacks[id(status)] = (status, success, error)
|
|
|
|
def emit_status(self, status_value: TimePixFlyStatus):
|
|
"""Resolve tracked status objects with a simulated backend state."""
|
|
self.status = status_value
|
|
for cb_id, (status, success, error) in list(self._status_callbacks.items()):
|
|
with status._lock:
|
|
if status.done:
|
|
self._status_callbacks.pop(cb_id, None)
|
|
continue
|
|
if status_value in success:
|
|
status.set_finished()
|
|
self._status_callbacks.pop(cb_id, None)
|
|
elif status_value in error:
|
|
status.set_exception(RuntimeError(f"backend entered {status_value.value}"))
|
|
self._status_callbacks.pop(cb_id, None)
|
|
|
|
def get_net_addresses(self):
|
|
"""Return a deterministic backend raw-data address."""
|
|
return NetAddresses(
|
|
control="127.0.0.1:8452", address="127.0.0.1:8451", server="127.0.0.1:8080"
|
|
)
|
|
|
|
|
|
class FakeBackend:
|
|
"""Minimal backend double used to isolate Timepix from backend integration."""
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
self.hostname = kwargs.get("hostname") or "localhost"
|
|
self.socket_port = kwargs.get("socket_port", 9876)
|
|
self.timepix_fly_client = FakeBackendClient()
|
|
self.on_connected = mock.Mock()
|
|
self.on_stage = mock.Mock()
|
|
self.on_stop = mock.Mock()
|
|
self.on_destroy = mock.Mock()
|
|
self.add_callback = mock.Mock()
|
|
self._trigger_status = StatusBase()
|
|
self._trigger_status.set_finished()
|
|
self._complete_status = StatusBase()
|
|
self._complete_status.set_finished()
|
|
|
|
def on_trigger(self):
|
|
"""Return a backend-prepared trigger status."""
|
|
return self._trigger_status
|
|
|
|
def on_trigger_finished(self):
|
|
"""Return the status that resolves when acquisition is complete."""
|
|
return self._complete_status
|
|
|
|
def on_complete(self, status=None):
|
|
"""Return a backend completion status."""
|
|
if status is None:
|
|
return self._complete_status
|
|
status.set_finished()
|
|
return status
|
|
|
|
|
|
def _finished_status(device=None):
|
|
"""Create a finished status for mocked signal set operations."""
|
|
status = DeviceStatus(device=device)
|
|
status.set_finished()
|
|
return status
|
|
|
|
|
|
def _force_signal_value(signal, value):
|
|
"""Set a mocked PV-backed signal value, including read-only EPICS signals."""
|
|
if hasattr(signal, "_read_pv"):
|
|
signal._read_pv.mock_data = value
|
|
return
|
|
signal.put(value)
|
|
|
|
|
|
def _message_value(signal):
|
|
"""Extract the signal payload from a BEC message signal."""
|
|
msg = signal.get()
|
|
return msg.signals[signal.name]["value"]
|
|
|
|
|
|
def _scan_status_msg(
|
|
scan_name: str = "line_scan",
|
|
additional_scan_parameters: dict | None = None,
|
|
exp_time: float = 0.1,
|
|
frames_per_trigger: int = 2,
|
|
num_points: int = 3,
|
|
) -> ScanStatusMessage:
|
|
"""v4-style scan status message."""
|
|
v4_info = ScanServerScanInfo(
|
|
scan_name=scan_name,
|
|
scan_id="test",
|
|
scan_type="software_triggered",
|
|
num_points=num_points,
|
|
exp_time=exp_time,
|
|
frames_per_trigger=frames_per_trigger,
|
|
additional_scan_parameters=additional_scan_parameters or {},
|
|
)
|
|
scan_parameters = {"exp_time": exp_time, "frames_per_trigger": frames_per_trigger}
|
|
info = v4_info.model_dump(mode="python")
|
|
info["scan_parameters"] = scan_parameters
|
|
return ScanStatusMessage(
|
|
scan_id="test",
|
|
status="open",
|
|
scan_name=scan_name,
|
|
scan_type="software_triggered",
|
|
num_points=num_points,
|
|
scan_parameters=scan_parameters,
|
|
info=info,
|
|
)
|
|
|
|
|
|
TIMEPIX_SCAN_PARAMETERS = {
|
|
"time_roi_step": 1,
|
|
"time_roi_n": 4,
|
|
"idx_off_start": 0,
|
|
"idx_off_stop": 2,
|
|
"idx_on_start": 2,
|
|
"idx_on_stop": 4,
|
|
}
|
|
|
|
LEVEL_SIGNALS = (
|
|
"xes_level_1_off",
|
|
"xes_level_1_on",
|
|
"xes_level_2_off",
|
|
"xes_level_2_on",
|
|
"xes_diff_1_on_off",
|
|
"xes_diff_2_on_off",
|
|
)
|
|
|
|
|
|
def _set_scan_parameters(dev, scan_name: str, additional_scan_parameters: dict | None = None):
|
|
"""Set scan parameters as on_stage does, without staging."""
|
|
dev.scan_info.msg = _scan_status_msg(scan_name, additional_scan_parameters)
|
|
dev.scan_parameters = fetch_scan_info(dev.scan_info)
|
|
|
|
|
|
def _xes_frames(spectra: np.ndarray):
|
|
"""StartFrame, one XesData frame and EndFrame for a (TRoiN, NumEnergyPoints) array."""
|
|
troin, n_energy_points = spectra.shape
|
|
start_frame = {
|
|
"type": "StartFrame",
|
|
"Mode": "TOA",
|
|
"TRoiStart": 0,
|
|
"TRoiStep": 1,
|
|
"TRoiN": troin,
|
|
"NumEnergyPoints": n_energy_points,
|
|
"save_interval": 1,
|
|
}
|
|
# energy points vary fastest
|
|
data_frames = [
|
|
{
|
|
"type": "XesData",
|
|
"period": 131000,
|
|
"totalEvents": int(spectra.sum()),
|
|
"TDSpectra": spectra.ravel().tolist(),
|
|
"beforeROI": 0,
|
|
"afterROI": 0,
|
|
}
|
|
]
|
|
end_frame = {"type": "EndFrame", "error": "", "periods": 4}
|
|
return start_frame, data_frames, end_frame
|
|
|
|
|
|
def _level_test_spectra() -> np.ndarray:
|
|
"""4 time bins x 8 energy points; groups sum to [4, 8, 12, 16] and [40, 80, 120, 160]."""
|
|
time_bin = np.arange(1, 5, dtype=np.float32)[:, None]
|
|
return np.hstack([np.repeat(time_bin, 4, axis=1), np.repeat(10 * time_bin, 4, axis=1)])
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def timepix():
|
|
"""Timepix device with mocked EPICS signals and a fully mocked backend."""
|
|
backend = FakeBackend()
|
|
scan_info = ScanInfo(msg=_scan_status_msg())
|
|
with (
|
|
mock.patch.object(ophyd, "cl") as mock_cl,
|
|
mock.patch("superxas_bec.devices.timepix.timepix.TimepixFlyBackend", return_value=backend),
|
|
# the default pixel map is slow to build
|
|
mock.patch(
|
|
"superxas_bec.devices.timepix.timepix.create_single_energy_per_chip_pixel_map",
|
|
return_value=PixelMap(chips=[[{"i": 0, "p": 0, "f": [1.0]}]]),
|
|
),
|
|
):
|
|
mock_cl.get_pv = MockPV
|
|
mock_cl.thread_class = threading.Thread
|
|
dev = Timepix(
|
|
name="timepix",
|
|
prefix="X10DA-ES-TPX1:",
|
|
backend_rest_url="localhost:8452",
|
|
hostname="localhost",
|
|
socket_port=9876,
|
|
scan_info=scan_info,
|
|
device_manager=DMMock(),
|
|
)
|
|
patch_dual_pvs(dev)
|
|
dev.backend = backend
|
|
dev._poll_thread = mock.Mock()
|
|
for walk in dev.walk_signals():
|
|
signal = walk.item
|
|
if hasattr(signal, "set") and hasattr(signal, "put"):
|
|
signal.set = mock.Mock(side_effect=lambda value, _sig=signal, **_kw: (_force_signal_value(_sig, value), _finished_status(_sig))[1]) # type: ignore[method-assign]
|
|
yield dev
|
|
dev._poll_thread_kill_event.set()
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def pixel_map():
|
|
"""Small valid pixel map used for stage tests."""
|
|
return PixelMap(chips=[[{"i": 0, "p": 0, "f": [1.0]}]])
|
|
|
|
|
|
def test_timepix_on_connected_configures_signals_and_registers_callback(timepix):
|
|
"""Connected hook should configure camera, file writer, and backend callback."""
|
|
timepix.cam.tdc2_enable.put(1)
|
|
timepix.on_connected()
|
|
|
|
assert timepix.cam.tdc1_enable.get() == 1
|
|
assert timepix.cam.tdc2_enable.get() == 0
|
|
assert timepix.cam.raw_enable.get() == 1
|
|
assert timepix.cam.array_counter.get() == 0
|
|
assert timepix.hdf.enable.get() == "1"
|
|
assert timepix.hdf.file_write_mode.get() == FILEWRITEMODE.STREAM.value
|
|
assert timepix.hdf.auto_save.get() == 1
|
|
assert timepix.cam.array_callbacks.get() == 1
|
|
timepix.backend.add_callback.assert_called_once_with(timepix.msg_buffer_callback)
|
|
timepix._poll_thread.start.assert_called_once()
|
|
|
|
|
|
def test_timepix_on_stage_configures_camera_writer_and_mocked_backend(timepix, pixel_map):
|
|
"""Stage should configure camera settings and forward config to the backend."""
|
|
timepix._pixel_map = pixel_map
|
|
|
|
with (
|
|
mock.patch(
|
|
"superxas_bec.devices.timepix.timepix.get_full_path",
|
|
return_value="/tmp/timepix/test_scan.h5",
|
|
),
|
|
):
|
|
_force_signal_value(timepix.hdf.enable, "Enable")
|
|
timepix.on_stage()
|
|
|
|
assert timepix.cam.acquire_time.get() == pytest.approx(0.1 - timepix._readout_time)
|
|
assert timepix.cam.acquire_period.get() == pytest.approx(0.1)
|
|
assert timepix.cam.num_images.get() == 2
|
|
assert timepix.cam.data_source.get() == DATASOURCE.IMAGE
|
|
assert timepix.hdf.file_path.get() == "/tmp/timepix"
|
|
assert timepix.hdf.file_name.get() == "test_scan.h5"
|
|
assert timepix.hdf.num_capture.get() == 6
|
|
assert timepix.hdf.capture.get() == 1
|
|
assert timepix.cam.raw_file_template.get() == ""
|
|
assert timepix.cam.raw_file_path.get() == "tcp://connect@127.0.0.1:8451"
|
|
assert timepix.backend.on_stage.call_count == 1
|
|
other_config = timepix.backend.on_stage.call_args.kwargs["other_config"]
|
|
assert (
|
|
other_config.output_uri
|
|
== "redis://localhost:0000/public/timepix_fly_backend/raw_data?scan-id=test"
|
|
)
|
|
assert other_config.TRoiStep == timepix.troistep
|
|
assert other_config.TRoiN == timepix.troin
|
|
assert timepix.backend.on_stage.call_args.kwargs["pixel_map"] == pixel_map
|
|
file_event = timepix.file_event.get()
|
|
assert file_event.file_path == "/tmp/timepix/test_scan.h5"
|
|
assert file_event.done is False
|
|
assert file_event.successful is False
|
|
|
|
|
|
def test_timepix_on_stage_configures_mocked_backend(timepix, pixel_map):
|
|
"""Stage should configure camera settings and forward config to the backend."""
|
|
timepix._pixel_map = pixel_map
|
|
|
|
with (
|
|
mock.patch(
|
|
"superxas_bec.devices.timepix.timepix.get_full_path",
|
|
return_value="/tmp/timepix/test_scan.h5",
|
|
),
|
|
):
|
|
_force_signal_value(timepix.hdf.enable, "Disable")
|
|
# File event should not be emitted when hdf.enable is "Disable"
|
|
with mock.patch.object(timepix.file_event, "put") as mock_file_event_put:
|
|
timepix.hdf.capture.put(0)
|
|
timepix.on_stage()
|
|
mock_file_event_put.assert_not_called()
|
|
|
|
assert timepix.hdf.capture.get() == 0
|
|
assert timepix.cam.acquire_time.get() == pytest.approx(0.1 - timepix._readout_time)
|
|
assert timepix.cam.acquire_period.get() == pytest.approx(0.1)
|
|
assert timepix.cam.num_images.get() == 2
|
|
assert timepix.cam.data_source.get() == DATASOURCE.IMAGE
|
|
assert timepix.cam.raw_file_template.get() == ""
|
|
assert timepix.cam.raw_file_path.get() == "tcp://connect@127.0.0.1:8451"
|
|
assert timepix.backend.on_stage.call_count == 1
|
|
other_config = timepix.backend.on_stage.call_args.kwargs["other_config"]
|
|
assert (
|
|
other_config.output_uri
|
|
== "redis://localhost:0000/public/timepix_fly_backend/raw_data?scan-id=test"
|
|
)
|
|
assert other_config.TRoiStep == timepix.troistep
|
|
assert other_config.TRoiN == timepix.troin
|
|
assert timepix.backend.on_stage.call_args.kwargs["pixel_map"] == pixel_map
|
|
|
|
|
|
def test_timepix_on_stage_without_xes_skips_backend_configuration(timepix):
|
|
"""When XES is disabled, the backend-specific stage call should be skipped."""
|
|
timepix.enable_xes = False
|
|
|
|
with mock.patch(
|
|
"superxas_bec.devices.timepix.timepix.get_full_path",
|
|
return_value="/tmp/timepix/no_xes_scan.h5",
|
|
):
|
|
timepix.on_stage()
|
|
|
|
timepix.backend.on_stage.assert_not_called()
|
|
|
|
|
|
def test_timepix_on_trigger_combines_camera_and_backend_status(timepix):
|
|
"""Trigger should arm the backend first and then drive the camera."""
|
|
backend_done = StatusBase()
|
|
backend_done.set_finished()
|
|
backend_finished = StatusBase()
|
|
timepix.backend._trigger_status = backend_done
|
|
timepix.backend._complete_status = backend_finished
|
|
|
|
status = timepix.on_trigger()
|
|
|
|
assert isinstance(status, StatusBase)
|
|
assert timepix.cam.acquire.get() == 1
|
|
_force_signal_value(timepix.cam.acquire_busy, ACQUIRESTATUS.ACQUIRING)
|
|
_force_signal_value(timepix.cam.acquire_busy, ACQUIRESTATUS.DONE)
|
|
backend_finished.set_finished()
|
|
status.wait(timeout=0.1)
|
|
assert status.done is True
|
|
assert status.success is True
|
|
|
|
|
|
def test_timepix_on_complete(timepix):
|
|
"""Complete should wait for writer/backend completion and emit a success file event."""
|
|
|
|
# I. Case I. With hdf.enable "Enable", should wait for both writer and camera to be done before emitting file event
|
|
timepix.hdf.enable.put("Enable")
|
|
timepix._full_path = "/tmp/timepix/final_scan.h5"
|
|
timepix._n_images = 3
|
|
_force_signal_value(timepix.cam.acquire_busy, ACQUIRESTATUS.ACQUIRING)
|
|
timepix.hdf.capture.put(ACQUIRESTATUS.ACQUIRING)
|
|
timepix.hdf.write_file.put(ACQUIRESTATUS.ACQUIRING)
|
|
|
|
status = timepix.on_complete()
|
|
|
|
_force_signal_value(timepix.cam.acquire_busy, ACQUIRESTATUS.DONE)
|
|
timepix.hdf.capture.put(ACQUIRESTATUS.DONE)
|
|
timepix.hdf.write_file.put(ACQUIRESTATUS.DONE)
|
|
timepix.hdf.write_status.put(0)
|
|
_force_signal_value(timepix.hdf.num_captured, 3)
|
|
status.wait(timeout=1)
|
|
|
|
file_event = timepix.file_event.get()
|
|
assert file_event.file_path == "/tmp/timepix/final_scan.h5"
|
|
assert file_event.done is True
|
|
assert file_event.successful is True
|
|
|
|
# II. Case II. With hdf.enable "Disable", should not wait for writer and only wait for camera to be done before emitting file event
|
|
timepix.hdf.enable.put("Disable")
|
|
timepix._n_images = 3
|
|
_force_signal_value(timepix.cam.acquire_busy, ACQUIRESTATUS.ACQUIRING)
|
|
timepix.hdf.capture.put(ACQUIRESTATUS.ACQUIRING)
|
|
timepix.hdf.write_file.put(ACQUIRESTATUS.ACQUIRING)
|
|
with mock.patch.object(timepix.file_event, "put") as mock_file_event_put:
|
|
status = timepix.on_complete()
|
|
_force_signal_value(timepix.cam.acquire_busy, ACQUIRESTATUS.DONE)
|
|
status.wait(timeout=1)
|
|
mock_file_event_put.assert_not_called()
|
|
assert status.done is True
|
|
assert status.success is True
|
|
|
|
|
|
def test_timepix_msg_buffer_callback_updates_xes_signals(timepix):
|
|
"""The backend message callback should populate all exported XES data signals."""
|
|
start_frame = {
|
|
"type": "StartFrame",
|
|
"Mode": "TOA",
|
|
"TRoiStart": 0,
|
|
"TRoiStep": 1,
|
|
"TRoiN": 2,
|
|
"NumEnergyPoints": 8,
|
|
"save_interval": 1,
|
|
}
|
|
data_frames = [
|
|
{
|
|
"type": "XesData",
|
|
"period": 131000,
|
|
"totalEvents": 36,
|
|
"TDSpectra": list(range(16)),
|
|
"beforeROI": 0,
|
|
"afterROI": 0,
|
|
}
|
|
]
|
|
end_frame = {"type": "EndFrame", "error": "", "periods": 4}
|
|
timepix.troin = 2
|
|
|
|
timepix.msg_buffer_callback(start_frame, data_frames, end_frame)
|
|
|
|
expected_xes = np.array(
|
|
[[0, 1, 2, 3, 4, 5, 6, 7], [8, 9, 10, 11, 12, 13, 14, 15]], dtype=np.float32
|
|
)
|
|
np.testing.assert_array_equal(_message_value(timepix.xes_data), expected_xes)
|
|
np.testing.assert_array_equal(
|
|
_message_value(timepix.xes_spectra), np.array([28, 92], dtype=np.float32)
|
|
)
|
|
np.testing.assert_array_equal(
|
|
_message_value(timepix.xes_energy_1), np.array([6, 38], dtype=np.float32)
|
|
)
|
|
np.testing.assert_array_equal(
|
|
_message_value(timepix.xes_energy_2), np.array([22, 54], dtype=np.float32)
|
|
)
|
|
assert _message_value(timepix.total_periods) == pytest.approx(4 / 131000)
|
|
np.testing.assert_array_equal(_message_value(timepix.tds_period), np.array([1.0]))
|
|
assert _message_value(timepix.total_events) == 36
|
|
|
|
|
|
def test_timepix_msg_buffer_callback_accumulates_xes_energy_per_scan(timepix):
|
|
"""Accumulated signals sum frames per scan; staging resets them."""
|
|
timepix.troin = 4
|
|
# one frame: xes_energy_1 = [4, 8, 12, 16], xes_energy_2 = [40, 80, 120, 160]
|
|
frames = _xes_frames(_level_test_spectra())
|
|
|
|
timepix.msg_buffer_callback(*frames)
|
|
timepix.msg_buffer_callback(*frames)
|
|
|
|
np.testing.assert_array_equal(
|
|
_message_value(timepix.xes_data_accumulated_1), np.array([8, 16, 24, 32])
|
|
)
|
|
np.testing.assert_array_equal(
|
|
_message_value(timepix.xes_data_accumulated_2), np.array([80, 160, 240, 320])
|
|
)
|
|
assert timepix.xes_data_accumulated_1.get().metadata["async_update"]["type"] == "replace"
|
|
# the per-frame signal still carries only the last frame
|
|
np.testing.assert_array_equal(_message_value(timepix.xes_energy_1), np.array([4, 8, 12, 16]))
|
|
|
|
_stage_for(timepix, "line_scan")
|
|
timepix.msg_buffer_callback(*frames)
|
|
|
|
np.testing.assert_array_equal(
|
|
_message_value(timepix.xes_data_accumulated_1), np.array([4, 8, 12, 16])
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("scan_name", [None, "line_scan"])
|
|
def test_timepix_msg_buffer_callback_skips_levels_outside_timepix_scan(timepix, scan_name):
|
|
"""On/off levels are untouched outside a timepix_scan."""
|
|
if scan_name is not None:
|
|
_set_scan_parameters(timepix, scan_name)
|
|
assert timepix._is_timepix_scan() is False
|
|
timepix.troin = 4
|
|
with ExitStack() as stack:
|
|
level_puts = [
|
|
stack.enter_context(mock.patch.object(getattr(timepix, name), "put"))
|
|
for name in LEVEL_SIGNALS
|
|
]
|
|
timepix.msg_buffer_callback(*_xes_frames(_level_test_spectra()))
|
|
|
|
for put in level_puts:
|
|
put.assert_not_called()
|
|
np.testing.assert_array_equal(
|
|
_message_value(timepix.xes_energy_1), np.array([4, 8, 12, 16], dtype=np.float32)
|
|
)
|
|
|
|
|
|
def test_timepix_msg_buffer_callback_publishes_on_off_levels(timepix):
|
|
"""Level/diff signals are the window means during a timepix_scan."""
|
|
_set_scan_parameters(timepix, "timepix_scan", TIMEPIX_SCAN_PARAMETERS)
|
|
timepix.troin = 4
|
|
|
|
timepix.msg_buffer_callback(*_xes_frames(_level_test_spectra()))
|
|
|
|
# xes_energy_1 = [4, 8, 12, 16], xes_energy_2 = [40, 80, 120, 160]
|
|
# off window [0, 2), on window [2, 4)
|
|
expected = {
|
|
"xes_level_1_off": 6.0,
|
|
"xes_level_1_on": 14.0,
|
|
"xes_diff_1_on_off": 8.0,
|
|
"xes_level_2_off": 60.0,
|
|
"xes_level_2_on": 140.0,
|
|
"xes_diff_2_on_off": 80.0,
|
|
}
|
|
for name, value in expected.items():
|
|
assert _message_value(getattr(timepix, name)) == pytest.approx(value), name
|
|
|
|
|
|
def test_timepix_on_stop_stops_camera_writer_and_backend(timepix):
|
|
"""Stop should stop camera acquisition and delegate backend stop."""
|
|
timepix.cam.acquire.put(1)
|
|
timepix.hdf.capture.put(1)
|
|
|
|
timepix.on_stop()
|
|
|
|
assert timepix.cam.acquire.get() == 0
|
|
assert timepix.hdf.capture.get() == 0
|
|
timepix.backend.on_stop.assert_called_once()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"window, expected",
|
|
[
|
|
(slice(0, 2), 1.5),
|
|
(slice(1, 4), 3.0),
|
|
(slice(2, 2), np.nan), # empty
|
|
(slice(3, 2), np.nan), # start after stop
|
|
(slice(2, 5), np.nan), # stop past the end
|
|
(slice(4, 5), np.nan), # start past the end
|
|
(slice(-1, 2), np.nan), # negative start
|
|
],
|
|
)
|
|
def test_window_mean(window, expected):
|
|
data = np.array([1.0, 2.0, 3.0, 4.0], dtype=np.float32)
|
|
np.testing.assert_equal(window_mean(data, window), expected)
|
|
|
|
|
|
def test_timepix_on_off_levels_are_nan_for_window_outside_received_bins(timepix):
|
|
"""Too few time bins give NaN levels."""
|
|
_set_scan_parameters(timepix, "timepix_scan", TIMEPIX_SCAN_PARAMETERS)
|
|
timepix.troin = 4
|
|
# only 3 time bins: xes_energy_1 = [4, 8, 12], xes_energy_2 = [40, 80, 120]
|
|
frames = _xes_frames(_level_test_spectra()[:3])
|
|
|
|
with mock.patch("superxas_bec.devices.timepix.timepix.logger") as logger:
|
|
timepix.msg_buffer_callback(*frames)
|
|
timepix.msg_buffer_callback(*frames)
|
|
|
|
assert _message_value(timepix.xes_level_1_off) == pytest.approx(6.0)
|
|
assert _message_value(timepix.xes_level_2_off) == pytest.approx(60.0)
|
|
for name in ("xes_level_1_on", "xes_level_2_on", "xes_diff_1_on_off", "xes_diff_2_on_off"):
|
|
assert np.isnan(_message_value(getattr(timepix, name))), name
|
|
# one warning per scan
|
|
assert logger.warning.call_count == 1
|
|
assert "on [2, 4)" in logger.warning.call_args.args[0]
|
|
|
|
|
|
def _stage_for(dev, scan_name: str, additional_scan_parameters: dict | None = None, **kwargs):
|
|
"""Stage via stage(), as the device server does."""
|
|
dev.scan_info.msg = _scan_status_msg(scan_name, additional_scan_parameters, **kwargs)
|
|
with mock.patch(
|
|
"superxas_bec.devices.timepix.timepix.get_full_path",
|
|
return_value="/tmp/timepix/test_scan.h5",
|
|
):
|
|
return dev.stage()
|
|
|
|
|
|
def _prepare_settings_before_timepix_scan(dev):
|
|
"""Image writing on, XES off, default time binning."""
|
|
_force_signal_value(dev.hdf.enable, "Enable")
|
|
dev.enable_xes = False
|
|
dev.troistep = 1
|
|
dev.troin = 5000
|
|
|
|
|
|
def _assert_settings_before_timepix_scan(dev):
|
|
assert dev.hdf.enable.get() == "Enable"
|
|
assert dev.enable_xes is False
|
|
assert dev.cam.tdc1_enable.get() == 0
|
|
assert dev.troistep == 1
|
|
assert dev.troin == 5000
|
|
assert dev._settings_before_timepix_scan is None
|
|
|
|
|
|
def test_timepix_scan_stage_overrides_settings_and_unstage_restores_them(timepix, pixel_map):
|
|
"""timepix_scan turns image writing off and XES on, for the scan only."""
|
|
timepix._pixel_map = pixel_map
|
|
_prepare_settings_before_timepix_scan(timepix)
|
|
params = {**TIMEPIX_SCAN_PARAMETERS, "time_roi_step": 2}
|
|
|
|
with mock.patch.object(timepix.file_event, "put") as file_event_put:
|
|
_stage_for(timepix, "timepix_scan", params)
|
|
file_event_put.assert_not_called()
|
|
|
|
assert timepix._staged == Staged.yes
|
|
assert timepix.hdf.enable.get() != "Enable"
|
|
assert timepix.hdf.capture.get() != 1
|
|
assert timepix.enable_xes is True
|
|
assert timepix.cam.tdc1_enable.get() == 1
|
|
other_config = timepix.backend.on_stage.call_args.kwargs["other_config"]
|
|
assert (other_config.TRoiStep, other_config.TRoiN) == (2, 4)
|
|
|
|
timepix.unstage()
|
|
|
|
assert timepix._staged == Staged.no
|
|
_assert_settings_before_timepix_scan(timepix)
|
|
|
|
|
|
def test_timepix_unstage_after_other_scan_keeps_settings(timepix):
|
|
"""Only timepix_scan touches image writing and XES."""
|
|
_prepare_settings_before_timepix_scan(timepix)
|
|
|
|
_stage_for(timepix, "line_scan")
|
|
timepix.unstage()
|
|
|
|
_assert_settings_before_timepix_scan(timepix)
|
|
|
|
|
|
def test_timepix_stop_restores_settings_of_aborted_timepix_scan(timepix, pixel_map):
|
|
"""Stop restores settings; a later unstage does not restore again."""
|
|
timepix._pixel_map = pixel_map
|
|
_prepare_settings_before_timepix_scan(timepix)
|
|
_stage_for(timepix, "timepix_scan", TIMEPIX_SCAN_PARAMETERS)
|
|
|
|
timepix.stop()
|
|
|
|
timepix.backend.on_stop.assert_called_once()
|
|
_assert_settings_before_timepix_scan(timepix)
|
|
|
|
# a user change after the abort survives the next unstage
|
|
timepix.set_enable_xes(True)
|
|
timepix.unstage()
|
|
assert timepix.enable_xes is True
|
|
|
|
|
|
def test_timepix_failed_restore_on_stop_is_retried_on_unstage(timepix, pixel_map):
|
|
timepix._pixel_map = pixel_map
|
|
_prepare_settings_before_timepix_scan(timepix)
|
|
_stage_for(timepix, "timepix_scan", TIMEPIX_SCAN_PARAMETERS)
|
|
working_set = timepix.hdf.enable.set
|
|
timepix.hdf.enable.set = mock.Mock(side_effect=TimeoutError("IOC did not answer"))
|
|
|
|
timepix.stop() # logs instead of raising
|
|
|
|
assert timepix._settings_before_timepix_scan is not None
|
|
timepix.hdf.enable.set = working_set
|
|
timepix.unstage()
|
|
_assert_settings_before_timepix_scan(timepix)
|
|
|
|
|
|
def test_timepix_failed_stage_restores_timepix_scan_settings(timepix, pixel_map):
|
|
"""A failing on_stage undoes the overrides."""
|
|
timepix._pixel_map = pixel_map
|
|
_prepare_settings_before_timepix_scan(timepix)
|
|
|
|
with pytest.raises(ValueError, match="Exposure time"):
|
|
_stage_for(timepix, "timepix_scan", TIMEPIX_SCAN_PARAMETERS, exp_time=0.001)
|
|
|
|
assert timepix._staged == Staged.no
|
|
_assert_settings_before_timepix_scan(timepix)
|
|
|
|
|
|
def test_timepix_failed_stage_and_failed_restore_keep_stage_error(timepix, pixel_map):
|
|
"""A failing restore does not hide the stage error; the next stage restores."""
|
|
timepix._pixel_map = pixel_map
|
|
_prepare_settings_before_timepix_scan(timepix)
|
|
working_set = timepix.hdf.enable.set
|
|
|
|
def set_or_time_out(value, **kwargs):
|
|
if value == "Enable": # only the restore writes this
|
|
raise TimeoutError("IOC did not answer")
|
|
return working_set(value, **kwargs)
|
|
|
|
timepix.hdf.enable.set = mock.Mock(side_effect=set_or_time_out)
|
|
|
|
with mock.patch("superxas_bec.devices.timepix.timepix.logger") as logger:
|
|
with pytest.raises(ValueError, match="Exposure time"):
|
|
_stage_for(timepix, "timepix_scan", TIMEPIX_SCAN_PARAMETERS, exp_time=0.001)
|
|
|
|
assert timepix._staged == Staged.no
|
|
assert timepix._settings_before_timepix_scan is not None
|
|
assert "IOC did not answer" in logger.error.call_args.args[0]
|
|
|
|
# not staged, so the next stage restores
|
|
timepix.hdf.enable.set = working_set
|
|
_stage_for(timepix, "line_scan")
|
|
|
|
_assert_settings_before_timepix_scan(timepix)
|
|
timepix.backend.on_stage.assert_not_called()
|
|
|
|
|
|
def test_timepix_on_destroy_cleans_up_backend(timepix):
|
|
"""Destroy should stop polling and forward cleanup to the backend."""
|
|
timepix.on_destroy()
|
|
|
|
assert timepix._poll_thread_kill_event.is_set() is True
|
|
timepix.backend.on_stop.assert_called_once()
|
|
timepix.backend.on_destroy.assert_called_once()
|
|
|
|
|
|
def test_timepix_on_prescan_returns_correct_status(timepix):
|
|
"""Prescan should return a finished status if the backend is ready."""
|
|
# Case I. With hdf.enable "Enable", should wait for both writer and camera to be ready
|
|
timepix.hdf.enable._read_pv.mock_data = "Enable"
|
|
timepix.cam.acquire_busy._read_pv.mock_data = ACQUIRESTATUS.ACQUIRING
|
|
timepix.hdf.capture._read_pv.mock_data = ACQUIRESTATUS.DONE
|
|
# Should be combined status of writer and control
|
|
status = timepix.on_pre_scan()
|
|
assert status.done is False
|
|
timepix.cam.acquire_busy._read_pv.mock_data = ACQUIRESTATUS.DONE
|
|
assert status.done is False
|
|
timepix.hdf.capture._read_pv.mock_data = ACQUIRESTATUS.ACQUIRING
|
|
status.wait(timeout=1)
|
|
assert status.done is True
|
|
assert status.success is True
|
|
|
|
# Case II. With hdf.enable "Enable", should wait for both writer and camera to be ready
|
|
timepix.hdf.enable._read_pv.mock_data = "Disable"
|
|
timepix.cam.acquire_busy._read_pv.mock_data = ACQUIRESTATUS.ACQUIRING
|
|
status = timepix.on_pre_scan()
|
|
assert status.done is False
|
|
timepix.cam.acquire_busy._read_pv.mock_data = ACQUIRESTATUS.DONE
|
|
status.wait(timeout=1)
|
|
assert status.done is True
|
|
assert status.success is True
|
|
|
|
|
|
# lost acquisitions
|
|
|
|
|
|
def _wait_for(condition, timeout: float = 2.0):
|
|
"""Poll until condition() is true."""
|
|
deadline = time.monotonic() + timeout
|
|
while not condition():
|
|
assert time.monotonic() < deadline, "condition not met in time"
|
|
time.sleep(0.01)
|
|
|
|
|
|
def _camera_acquires(dev):
|
|
"""Simulate one camera acquisition."""
|
|
_force_signal_value(dev.cam.acquire_busy, ACQUIRESTATUS.ACQUIRING)
|
|
_force_signal_value(dev.cam.acquire_busy, ACQUIRESTATUS.DONE)
|
|
|
|
|
|
def _empty_frames():
|
|
"""Frames without any TDC period."""
|
|
start_frame, data_frames, end_frame = _xes_frames(np.zeros((4, 8), dtype=np.float32))
|
|
data_frames[0]["period"] = 0
|
|
return start_frame, data_frames, {**end_frame, "periods": 0}
|
|
|
|
|
|
@pytest.fixture
|
|
def xes_timepix(timepix):
|
|
"""Timepix set up for XES: image writing off, 1 s exposure, 1 image."""
|
|
_force_signal_value(timepix.hdf.enable, "Disable")
|
|
timepix.troin = 4
|
|
timepix._exposure_per_trigger = 1.0
|
|
timepix._images_per_trigger = 1
|
|
backend_done = [StatusBase(), StatusBase()]
|
|
timepix.backend.on_trigger_finished = mock.Mock(side_effect=backend_done)
|
|
timepix.backend_done = backend_done
|
|
return timepix
|
|
|
|
|
|
def test_timepix_lost_acquisition_is_repeated_once(xes_timepix):
|
|
"""An early-closed frame is dropped and the acquisition repeated."""
|
|
timepix = xes_timepix
|
|
with (
|
|
mock.patch.object(timepix.xes_data, "put") as xes_put,
|
|
mock.patch.object(timepix.static_spectra, "put") as spectra_put,
|
|
# a 2 x 2 detector image
|
|
mock.patch.object(timepix.image.array_size.height, "get", return_value=2),
|
|
mock.patch.object(timepix.image.array_size.width, "get", return_value=2),
|
|
mock.patch.object(timepix.image.array_data, "get", return_value=np.arange(4)),
|
|
):
|
|
status = timepix.on_trigger()
|
|
assert timepix.cam.acquire.get() == 1
|
|
|
|
# frame closes at once: nothing is published
|
|
timepix.msg_buffer_callback(*_empty_frames())
|
|
xes_put.assert_not_called()
|
|
timepix._on_new_image_received(value=1, old_value=0)
|
|
spectra_put.assert_not_called()
|
|
|
|
_camera_acquires(timepix)
|
|
timepix.backend_done[0].set_finished()
|
|
assert status.done is False
|
|
|
|
_wait_for(lambda: timepix.backend.on_trigger_finished.call_count == 2)
|
|
timepix.msg_buffer_callback(*_xes_frames(_level_test_spectra()))
|
|
timepix._on_new_image_received(value=2, old_value=1)
|
|
_camera_acquires(timepix)
|
|
timepix.backend_done[1].set_finished()
|
|
status.wait(timeout=2)
|
|
|
|
assert status.success is True
|
|
xes_put.assert_called_once()
|
|
np.testing.assert_array_equal(xes_put.call_args.args[0], _level_test_spectra())
|
|
spectra_put.assert_called_once()
|
|
assert not timepix._xes_acquisitions
|
|
|
|
|
|
def test_timepix_lost_repetition_is_published(xes_timepix):
|
|
"""A lost repetition is published as is."""
|
|
timepix = xes_timepix
|
|
with mock.patch.object(timepix.xes_data, "put") as xes_put:
|
|
status = timepix.on_trigger()
|
|
timepix.msg_buffer_callback(*_empty_frames())
|
|
timepix._on_new_image_received(value=1, old_value=0)
|
|
_camera_acquires(timepix)
|
|
timepix.backend_done[0].set_finished()
|
|
_wait_for(lambda: timepix.backend.on_trigger_finished.call_count == 2)
|
|
|
|
with mock.patch("superxas_bec.devices.timepix.timepix.logger") as logger:
|
|
timepix.msg_buffer_callback(*_empty_frames())
|
|
_camera_acquires(timepix)
|
|
timepix.backend_done[1].set_finished()
|
|
status.wait(timeout=2)
|
|
|
|
assert status.success is True
|
|
xes_put.assert_called_once()
|
|
assert "cannot be repeated" in logger.error.call_args.args[0]
|
|
|
|
|
|
def test_timepix_frame_without_tdc_after_exposure_is_published(xes_timepix):
|
|
"""No TDC pulse over a full exposure is published, not repeated."""
|
|
timepix = xes_timepix
|
|
timepix._exposure_per_trigger = 0.0 # exposure already over
|
|
with (
|
|
mock.patch.object(timepix.xes_data, "put") as xes_put,
|
|
mock.patch("superxas_bec.devices.timepix.timepix.logger") as logger,
|
|
):
|
|
status = timepix.on_trigger()
|
|
timepix.msg_buffer_callback(*_empty_frames())
|
|
_camera_acquires(timepix)
|
|
timepix.backend_done[0].set_finished()
|
|
status.wait(timeout=1)
|
|
|
|
assert status.success is True
|
|
xes_put.assert_called_once()
|
|
assert timepix.backend.on_trigger_finished.call_count == 1
|
|
assert "no TDC pulse" in logger.error.call_args.args[0]
|
|
|
|
|
|
def test_timepix_lost_acquisition_not_repeated_while_writing_images(xes_timepix):
|
|
"""No repetition with HDF5 image writing on."""
|
|
timepix = xes_timepix
|
|
_force_signal_value(timepix.hdf.enable, "Enable")
|
|
with mock.patch.object(timepix.xes_data, "put") as xes_put:
|
|
status = timepix.on_trigger()
|
|
timepix.msg_buffer_callback(*_empty_frames())
|
|
_camera_acquires(timepix)
|
|
timepix.backend_done[0].set_finished()
|
|
status.wait(timeout=1)
|
|
|
|
assert status.success is True
|
|
xes_put.assert_called_once()
|
|
assert timepix.backend.on_trigger_finished.call_count == 1
|
|
|
|
|
|
def test_timepix_stop_cancels_repetition_of_lost_acquisition(xes_timepix):
|
|
"""A stop during a lost acquisition cancels the trigger without repeating."""
|
|
timepix = xes_timepix
|
|
status = timepix.on_trigger()
|
|
timepix.msg_buffer_callback(*_empty_frames())
|
|
|
|
timepix.stop()
|
|
|
|
with pytest.raises(DeviceStoppedError):
|
|
status.wait(timeout=1)
|
|
_camera_acquires(timepix)
|
|
time.sleep(0.2)
|
|
assert timepix.backend.on_trigger_finished.call_count == 1
|
|
assert not timepix._xes_acquisitions
|
|
assert timepix._images_to_skip == 0
|
|
|
|
|
|
def test_timepix_on_connected_leaves_detector_setup_to_the_ioc(timepix):
|
|
"""Trigger and TDC edge/output PVs belong to the IOC setup; connecting must not write them."""
|
|
timepix.cam.trigger_mode.put(TRIGGERMODE.EXTERNAL)
|
|
timepix.cam.exposure_mode.put(EXPOSUREMODE.TRIGGER_WIDTH)
|
|
timepix.cam.tdc1_edge.put(TDCEdge.FALLING)
|
|
timepix.cam.tdc1_output.put(TDCOuput.CHANNEL_0)
|
|
|
|
timepix.on_connected()
|
|
|
|
assert timepix.cam.trigger_mode.get() == TRIGGERMODE.EXTERNAL
|
|
assert timepix.cam.exposure_mode.get() == EXPOSUREMODE.TRIGGER_WIDTH
|
|
assert timepix.cam.tdc1_edge.get() == TDCEdge.FALLING
|
|
assert timepix.cam.tdc1_output.get() == TDCOuput.CHANNEL_0
|
|
|
|
|
|
def test_timepix_check_detector_setup_warns_about_unexpected_values(timepix):
|
|
"""A setup PV that differs from what BEC expects is reported, not overwritten."""
|
|
timepix._enable_xes = True
|
|
timepix.cam.trigger_mode.put(TRIGGERMODE.EXTERNAL)
|
|
timepix.cam.tdc1_edge.put(TDCEdge.FALLING)
|
|
|
|
with mock.patch("superxas_bec.devices.timepix.timepix.logger") as logger:
|
|
timepix._check_detector_setup()
|
|
|
|
message = logger.warning.call_args.args[0]
|
|
assert "trigger_mode" in message and "tdc1_edge" in message
|
|
assert timepix.cam.trigger_mode.get() == TRIGGERMODE.EXTERNAL
|
|
|
|
|
|
def test_timepix_check_detector_setup_is_quiet_when_setup_matches(timepix):
|
|
"""No warning when the IOC setup matches."""
|
|
timepix._enable_xes = True
|
|
with mock.patch("superxas_bec.devices.timepix.timepix.logger") as logger:
|
|
timepix._check_detector_setup()
|
|
logger.warning.assert_not_called()
|