wi add test timepix_fly client
This commit is contained in:
@@ -1,7 +1,8 @@
|
||||
"""
|
||||
TimePix Detector class for interfacing with the TimePix detector. The timepix_signals module
|
||||
implements the HTTP communication to the REST API for the tpx3app app. The control interface
|
||||
is implemented via EPICS IOC.
|
||||
implements the HTTP communication to the REST API for the tpx3app app. The implementation
|
||||
of the backend is stored in the timepix_fly_client module. This is combined with the control
|
||||
interface in EPICS, which is implemented via the 'ASItpxCam' class.
|
||||
"""
|
||||
|
||||
import enum
|
||||
@@ -151,9 +152,9 @@ class Timepix(PSIDeviceBase, TimePixControl):
|
||||
prefix (str): EPICS prefix for the device.
|
||||
backend_rest_url (str): URL of the TimePixFly backend REST API.
|
||||
hostname (str | None): Hostname of the machine running the backend. Defaults to None
|
||||
which will use the current machine's hostname.
|
||||
which will use the current machine's hostname.
|
||||
socket_port (int): Port for the socket connection to the backend. Defaults to 0
|
||||
which will use the default port from the backend.
|
||||
which will use the default port from the backend.
|
||||
scan_info: Scan information object, if available.
|
||||
device_manager: Device manager instance, if available.
|
||||
**kwargs: Additional keyword arguments for the base class.
|
||||
@@ -297,6 +298,7 @@ class Timepix(PSIDeviceBase, TimePixControl):
|
||||
######################################################################
|
||||
### Beamline specific methods for the TimePix Detector integration ###
|
||||
######################################################################
|
||||
|
||||
def on_init(self) -> None:
|
||||
"""
|
||||
Called when the device is initialized.
|
||||
|
||||
@@ -75,6 +75,7 @@ class TimepixFlyBackend:
|
||||
self._data_thread: threading.Thread | None = None
|
||||
self._data_thread_shutdown_event = threading.Event()
|
||||
atexit.register(self.on_destroy) # Ensure cleanup on exit
|
||||
self.on_init() # TODO is this needed after registering the atexit handler?
|
||||
|
||||
###################################################
|
||||
###### Hooks for the PSIDeviceBase interface ######
|
||||
|
||||
@@ -183,7 +183,7 @@ class TimepixFlyClient:
|
||||
# pylint: disable=raise-missing-from
|
||||
def wait_for_connection(self, timeout: float = 6) -> None:
|
||||
"""
|
||||
Wait for the connection to the StdDAQ to be established.
|
||||
Wait for the connection to the TimepixFly client connection to be established.
|
||||
|
||||
Args:
|
||||
timeout (float): timeout for the request
|
||||
@@ -283,9 +283,7 @@ class TimepixFlyClient:
|
||||
self._started = False # Should this be made thread-safe?
|
||||
|
||||
def shutdown(self):
|
||||
"""
|
||||
Shutdown the StdDAQ client.
|
||||
"""
|
||||
"""Shutdown the TimepixFlyClient client."""
|
||||
self._shutdown_event.set()
|
||||
if self.ws_client is not None:
|
||||
self.ws_client.close()
|
||||
|
||||
@@ -1,139 +1,139 @@
|
||||
"""Utility module for the Timepix detector."""
|
||||
# """Utility module for the Timepix detector."""
|
||||
|
||||
from __future__ import annotations
|
||||
# from __future__ import annotations
|
||||
|
||||
from collections import defaultdict
|
||||
from typing import Dict, List, Tuple
|
||||
# from collections import defaultdict
|
||||
# from typing import Dict, List, Tuple
|
||||
|
||||
import numpy as np
|
||||
# import numpy as np
|
||||
|
||||
ROI = List[Tuple[float, float]]
|
||||
# ROI = List[Tuple[float, float]]
|
||||
|
||||
|
||||
def order_roi_corners_simple(roi: ROI) -> np.ndarray:
|
||||
"""Order ROI corners as [top-left, top-right, bottom-right, bottom-left]."""
|
||||
pts = np.array(roi, dtype=float)
|
||||
cx, cy = pts.mean(axis=0)
|
||||
angles = np.arctan2(pts[:, 1] - cy, pts[:, 0] - cx)
|
||||
idx = np.argsort(angles)
|
||||
ordered = pts[idx]
|
||||
# Ensure clockwise order
|
||||
if np.cross(ordered[1] - ordered[0], ordered[2] - ordered[0]) < 0:
|
||||
ordered = ordered[::-1]
|
||||
return ordered[:4]
|
||||
# def order_roi_corners_simple(roi: ROI) -> np.ndarray:
|
||||
# """Order ROI corners as [top-left, top-right, bottom-right, bottom-left]."""
|
||||
# pts = np.array(roi, dtype=float)
|
||||
# cx, cy = pts.mean(axis=0)
|
||||
# angles = np.arctan2(pts[:, 1] - cy, pts[:, 0] - cx)
|
||||
# idx = np.argsort(angles)
|
||||
# ordered = pts[idx]
|
||||
# # Ensure clockwise order
|
||||
# if np.cross(ordered[1] - ordered[0], ordered[2] - ordered[0]) < 0:
|
||||
# ordered = ordered[::-1]
|
||||
# return ordered[:4]
|
||||
|
||||
|
||||
def compute_affine_transform(
|
||||
src: np.ndarray, dst: np.ndarray, preserve_scale: bool = True
|
||||
) -> np.ndarray:
|
||||
"""Compute affine transform mapping src -> dst. Optionally preserve pixel scale."""
|
||||
if preserve_scale:
|
||||
# Solve for rotation+translation only
|
||||
A = np.array(
|
||||
[
|
||||
[src[0, 0], -src[0, 1], 1, 0],
|
||||
[src[0, 1], src[0, 0], 0, 1],
|
||||
[src[1, 0], -src[1, 1], 1, 0],
|
||||
[src[1, 1], src[1, 0], 0, 1],
|
||||
]
|
||||
)
|
||||
b = dst[:2].ravel()
|
||||
x, residuals, _, _ = np.linalg.lstsq(A, b, rcond=None)
|
||||
a, b_, tx, ty = x
|
||||
return np.array([[a, -b_, tx], [b_, a, ty]])
|
||||
else:
|
||||
# Full affine transform
|
||||
src_h = np.hstack([src, np.ones((4, 1))])
|
||||
dst_h = dst
|
||||
M, _, _, _ = np.linalg.lstsq(src_h, dst_h, rcond=None)
|
||||
return M.T
|
||||
# def compute_affine_transform(
|
||||
# src: np.ndarray, dst: np.ndarray, preserve_scale: bool = True
|
||||
# ) -> np.ndarray:
|
||||
# """Compute affine transform mapping src -> dst. Optionally preserve pixel scale."""
|
||||
# if preserve_scale:
|
||||
# # Solve for rotation+translation only
|
||||
# A = np.array(
|
||||
# [
|
||||
# [src[0, 0], -src[0, 1], 1, 0],
|
||||
# [src[0, 1], src[0, 0], 0, 1],
|
||||
# [src[1, 0], -src[1, 1], 1, 0],
|
||||
# [src[1, 1], src[1, 0], 0, 1],
|
||||
# ]
|
||||
# )
|
||||
# b = dst[:2].ravel()
|
||||
# x, residuals, _, _ = np.linalg.lstsq(A, b, rcond=None)
|
||||
# a, b_, tx, ty = x
|
||||
# return np.array([[a, -b_, tx], [b_, a, ty]])
|
||||
# else:
|
||||
# # Full affine transform
|
||||
# src_h = np.hstack([src, np.ones((4, 1))])
|
||||
# dst_h = dst
|
||||
# M, _, _, _ = np.linalg.lstsq(src_h, dst_h, rcond=None)
|
||||
# return M.T
|
||||
|
||||
|
||||
def apply_affine_transform(coords: np.ndarray, affine: np.ndarray) -> np.ndarray:
|
||||
"""Apply affine transform to coordinates."""
|
||||
coords_h = np.hstack([coords, np.ones((coords.shape[0], 1))])
|
||||
transformed = coords_h @ affine.T
|
||||
return transformed[:, :2]
|
||||
# def apply_affine_transform(coords: np.ndarray, affine: np.ndarray) -> np.ndarray:
|
||||
# """Apply affine transform to coordinates."""
|
||||
# coords_h = np.hstack([coords, np.ones((coords.shape[0], 1))])
|
||||
# transformed = coords_h @ affine.T
|
||||
# return transformed[:, :2]
|
||||
|
||||
|
||||
def roi_pixel_hits(
|
||||
image_shape: Tuple[int, int], roi: ROI, start_idx: int = 0, min_fraction_diff: float = 0.0
|
||||
) -> List[Dict]:
|
||||
"""
|
||||
For each ROI, return list of hits as dicts {'i': (x,y), 'p': row_idx, 'f': fraction}.
|
||||
Supports rotated rectangles using bilinear fraction splitting.
|
||||
"""
|
||||
hits_dict: Dict[Tuple[int, int], Dict[str, list]] = defaultdict(lambda: {"p": [], "f": []})
|
||||
# def roi_pixel_hits(
|
||||
# image_shape: Tuple[int, int], roi: ROI, start_idx: int = 0, min_fraction_diff: float = 0.0
|
||||
# ) -> List[Dict]:
|
||||
# """
|
||||
# For each ROI, return list of hits as dicts {'i': (x,y), 'p': row_idx, 'f': fraction}.
|
||||
# Supports rotated rectangles using bilinear fraction splitting.
|
||||
# """
|
||||
# hits_dict: Dict[Tuple[int, int], Dict[str, list]] = defaultdict(lambda: {"p": [], "f": []})
|
||||
|
||||
corners = order_roi_corners_simple(roi)
|
||||
height = int(np.linalg.norm(corners[0] - corners[3])) + 1
|
||||
width = int(np.linalg.norm(corners[0] - corners[1])) + 1
|
||||
# corners = order_roi_corners_simple(roi)
|
||||
# height = int(np.linalg.norm(corners[0] - corners[3])) + 1
|
||||
# width = int(np.linalg.norm(corners[0] - corners[1])) + 1
|
||||
|
||||
dst = np.array([[0, 0], [width - 1, 0], [width - 1, height - 1], [0, height - 1]], dtype=float)
|
||||
affine_mat = compute_affine_transform(corners, dst) # 2x3
|
||||
# dst = np.array([[0, 0], [width - 1, 0], [width - 1, height - 1], [0, height - 1]], dtype=float)
|
||||
# affine_mat = compute_affine_transform(corners, dst) # 2x3
|
||||
|
||||
# Bounding box in image
|
||||
min_x = max(int(np.floor(corners[:, 0].min())), 0)
|
||||
max_x = min(int(np.ceil(corners[:, 0].max())), image_shape[1] - 1)
|
||||
min_y = max(int(np.floor(corners[:, 1].min())), 0)
|
||||
max_y = min(int(np.ceil(corners[:, 1].max())), image_shape[0] - 1)
|
||||
# # Bounding box in image
|
||||
# min_x = max(int(np.floor(corners[:, 0].min())), 0)
|
||||
# max_x = min(int(np.ceil(corners[:, 0].max())), image_shape[1] - 1)
|
||||
# min_y = max(int(np.floor(corners[:, 1].min())), 0)
|
||||
# max_y = min(int(np.ceil(corners[:, 1].max())), image_shape[0] - 1)
|
||||
|
||||
yy, xx = np.meshgrid(np.arange(min_y, max_y + 1), np.arange(min_x, max_x + 1), indexing="ij")
|
||||
coords = np.stack([xx.ravel(), yy.ravel()], axis=1) # N x 2
|
||||
# yy, xx = np.meshgrid(np.arange(min_y, max_y + 1), np.arange(min_x, max_x + 1), indexing="ij")
|
||||
# coords = np.stack([xx.ravel(), yy.ravel()], axis=1) # N x 2
|
||||
|
||||
local_coords = apply_affine_transform(coords, affine_mat)
|
||||
x_local = local_coords[:, 0]
|
||||
y_local = local_coords[:, 1]
|
||||
# local_coords = apply_affine_transform(coords, affine_mat)
|
||||
# x_local = local_coords[:, 0]
|
||||
# y_local = local_coords[:, 1]
|
||||
|
||||
# Keep pixels inside ROI rectangle
|
||||
mask = (x_local >= 0) & (x_local <= width - 1) & (y_local >= 0) & (y_local <= height - 1)
|
||||
coords_in = coords[mask]
|
||||
x_in = x_local[mask]
|
||||
y_in = y_local[mask]
|
||||
# # Keep pixels inside ROI rectangle
|
||||
# mask = (x_local >= 0) & (x_local <= width - 1) & (y_local >= 0) & (y_local <= height - 1)
|
||||
# coords_in = coords[mask]
|
||||
# x_in = x_local[mask]
|
||||
# y_in = y_local[mask]
|
||||
|
||||
# Bilinear fractions
|
||||
x0 = np.floor(x_in).astype(int)
|
||||
y0 = np.floor(y_in).astype(int)
|
||||
dx = x_in - x0
|
||||
dy = y_in - y0
|
||||
# # Bilinear fractions
|
||||
# x0 = np.floor(x_in).astype(int)
|
||||
# y0 = np.floor(y_in).astype(int)
|
||||
# dx = x_in - x0
|
||||
# dy = y_in - y0
|
||||
|
||||
for coord, x0i, y0i, dxv, dyv in zip(coords_in, x0, y0, dx, dy):
|
||||
row_base = start_idx
|
||||
# for coord, x0i, y0i, dxv, dyv in zip(coords_in, x0, y0, dx, dy):
|
||||
# row_base = start_idx
|
||||
|
||||
# Contributions to 4 neighboring "rows"
|
||||
contributions = [
|
||||
(row_base + y0i, (1 - dxv) * (1 - dyv)),
|
||||
(row_base + y0i, dxv * (1 - dyv)),
|
||||
(row_base + y0i + 1, (1 - dxv) * dyv),
|
||||
(row_base + y0i + 1, dxv * dyv),
|
||||
]
|
||||
# # Contributions to 4 neighboring "rows"
|
||||
# contributions = [
|
||||
# (row_base + y0i, (1 - dxv) * (1 - dyv)),
|
||||
# (row_base + y0i, dxv * (1 - dyv)),
|
||||
# (row_base + y0i + 1, (1 - dxv) * dyv),
|
||||
# (row_base + y0i + 1, dxv * dyv),
|
||||
# ]
|
||||
|
||||
# Filter negligible contributions
|
||||
contributions = [(p, f) for p, f in contributions if f >= min_fraction_diff]
|
||||
# # Filter negligible contributions
|
||||
# contributions = [(p, f) for p, f in contributions if f >= min_fraction_diff]
|
||||
|
||||
# Normalize fractions to sum 1
|
||||
if contributions:
|
||||
total_f = sum(f for _, f in contributions)
|
||||
contributions = [(p, f / total_f) for p, f in contributions]
|
||||
# # Normalize fractions to sum 1
|
||||
# if contributions:
|
||||
# total_f = sum(f for _, f in contributions)
|
||||
# contributions = [(p, f / total_f) for p, f in contributions]
|
||||
|
||||
for p, f in contributions:
|
||||
key = (int(coord[0]), int(coord[1]))
|
||||
hits_dict[key]["p"].append(int(p))
|
||||
hits_dict[key]["f"].append(float(f))
|
||||
# for p, f in contributions:
|
||||
# key = (int(coord[0]), int(coord[1]))
|
||||
# hits_dict[key]["p"].append(int(p))
|
||||
# hits_dict[key]["f"].append(float(f))
|
||||
|
||||
hits_roi = [{"i": key, "p": value["p"], "f": value["f"]} for key, value in hits_dict.items()]
|
||||
return hits_roi
|
||||
# hits_roi = [{"i": key, "p": value["p"], "f": value["f"]} for key, value in hits_dict.items()]
|
||||
# return hits_roi
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
image_shape = (512, 512)
|
||||
rois = [
|
||||
[(25, 25), (50, 50), (50, 25), (75, 50)], # rotated 45 degrees rectangle
|
||||
[(300, 300), (400, 300), (400, 400), (300, 400)], # upright rectangle
|
||||
]
|
||||
# if __name__ == "__main__":
|
||||
# image_shape = (512, 512)
|
||||
# rois = [
|
||||
# [(25, 25), (50, 50), (50, 25), (75, 50)], # rotated 45 degrees rectangle
|
||||
# [(300, 300), (400, 300), (400, 400), (300, 400)], # upright rectangle
|
||||
# ]
|
||||
|
||||
hits_0 = roi_pixel_hits(image_shape, rois[0], min_fraction_diff=0.1)
|
||||
hits_1 = roi_pixel_hits(image_shape, rois[1], start_idx=10, min_fraction_diff=0.2)
|
||||
# hits_0 = roi_pixel_hits(image_shape, rois[0], min_fraction_diff=0.1)
|
||||
# hits_1 = roi_pixel_hits(image_shape, rois[1], start_idx=10, min_fraction_diff=0.2)
|
||||
|
||||
print(hits_0[:5])
|
||||
print(hits_1[:5])
|
||||
# print(hits_0[:5])
|
||||
# print(hits_1[:5])
|
||||
|
||||
@@ -0,0 +1,15 @@
|
||||
"""This module tests the Timepix Fly backend functionality."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_backend import TimepixFlyBackend
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def timepix_fly_backend():
|
||||
"""Fixture for creating a Timepix Fly backend instance."""
|
||||
backend = TimepixFlyBackend(backend_rest_url="http://localhost:8000")
|
||||
yield backend
|
||||
backend.on_destroy()
|
||||
@@ -0,0 +1,276 @@
|
||||
"""Module to test the Timepix Fly client functionality."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
from ophyd import StatusBase
|
||||
from websockets import State
|
||||
|
||||
from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_client import (
|
||||
TimepixFlyClient,
|
||||
TimePixFlyStatus,
|
||||
TimePixStatusError,
|
||||
)
|
||||
from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_interface import ProgramState
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def timepix_fly_client():
|
||||
"""Fixture for creating a Timepix Fly client instance."""
|
||||
client = TimepixFlyClient(rest_url="http://localhost:8000", ws_url="ws://localhost:8000/ws")
|
||||
try:
|
||||
yield client
|
||||
finally:
|
||||
client.shutdown()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"return_state",
|
||||
[
|
||||
ProgramState(state=TimePixFlyStatus.CONFIG),
|
||||
ProgramState(state=TimePixFlyStatus.COLLECT),
|
||||
ProgramState(state=TimePixFlyStatus.SETUP),
|
||||
],
|
||||
)
|
||||
def test_timepix_fly_client_stop_running_collection(timepix_fly_client, return_state):
|
||||
"""Test the on_connected method of the Timepix Fly client."""
|
||||
with (
|
||||
mock.patch.object(timepix_fly_client, "stop_collect") as mock_stop_collect,
|
||||
mock.patch.object(timepix_fly_client, "state", return_value=return_state),
|
||||
):
|
||||
timepix_fly_client.stop_running_collection()
|
||||
|
||||
if return_state.state == TimePixFlyStatus.CONFIG:
|
||||
assert mock_stop_collect.call_count == 0, "Stop collect should be called once."
|
||||
timepix_fly_client._started = True
|
||||
timepix_fly_client.stop_running_collection()
|
||||
assert mock_stop_collect.call_count == 1, "Stop collect should not be called again."
|
||||
else:
|
||||
assert mock_stop_collect.call_count == 1, "Stop collect should be called once."
|
||||
timepix_fly_client._started = True
|
||||
timepix_fly_client.stop_running_collection()
|
||||
assert mock_stop_collect.call_count == 2, "Stop collect should be called in CONFIG."
|
||||
|
||||
|
||||
def test_timepix_fly_client_on_connected(timepix_fly_client):
|
||||
"""
|
||||
Test timepix fly client connect method.
|
||||
|
||||
This simply ensures that all methods are called. They are tested separately.
|
||||
"""
|
||||
with (
|
||||
mock.patch.object(timepix_fly_client, "stop_running_collection") as mock_stop_collection,
|
||||
mock.patch.object(timepix_fly_client, "connect") as mock_connect,
|
||||
mock.patch.object(timepix_fly_client, "wait_for_connection") as mock_wait_for_connection,
|
||||
):
|
||||
timepix_fly_client.on_connected()
|
||||
mock_stop_collection.assert_called_once()
|
||||
mock_connect.assert_called_once()
|
||||
mock_wait_for_connection.assert_called_once()
|
||||
|
||||
|
||||
def test_timepix_fly_client_connect(timepix_fly_client):
|
||||
"""This tests the connect method of timepix fly client."""
|
||||
module_path = "superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_client"
|
||||
# Patch Thread so we don't actually start the background loop
|
||||
with mock.patch(f"{module_path}.threading.Thread") as mock_thread_cls:
|
||||
mock_thread = mock.Mock()
|
||||
mock_thread_cls.return_value = mock_thread
|
||||
|
||||
timepix_fly_client.connect()
|
||||
|
||||
# Thread should be created with the update loop as target and daemon True
|
||||
mock_thread_cls.assert_called_once()
|
||||
# start() must be called on the created thread
|
||||
mock_thread.start.assert_called_once()
|
||||
|
||||
|
||||
def test_timepix_fly_client_wait_for_connection(timepix_fly_client):
|
||||
"""This tests the wait_for_connection method of timepix fly client."""
|
||||
module_path = "superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_client"
|
||||
# Case 1: ws_client already present and OPEN
|
||||
mock_client = mock.Mock()
|
||||
type(mock_client).state = mock.PropertyMock(return_value=State.OPEN)
|
||||
timepix_fly_client.ws_client = mock_client
|
||||
|
||||
# Should return immediately
|
||||
timepix_fly_client.wait_for_connection(timeout=0.1)
|
||||
|
||||
# Case 2: connect() establishes the connection
|
||||
timepix_fly_client.ws_client = None
|
||||
mock_client2 = mock.Mock()
|
||||
type(mock_client2).state = mock.PropertyMock(return_value=State.OPEN)
|
||||
with mock.patch(f"{module_path}.connect", return_value=mock_client2) as mock_connect:
|
||||
timepix_fly_client.wait_for_connection(timeout=0.1)
|
||||
mock_connect.assert_called_once()
|
||||
|
||||
|
||||
def test_timepix_fly_client_ws_send_and_received(timepix_fly_client):
|
||||
"""This tests the _ws_send_and_receive method of timepix fly client."""
|
||||
# Prepare a mock ws_client where recv returns a message
|
||||
mock_ws = mock.Mock()
|
||||
mock_ws.recv.return_value = "init"
|
||||
timepix_fly_client.ws_client = mock_ws
|
||||
|
||||
# Patch _on_received_ws_message to ensure it's invoked
|
||||
with mock.patch.object(timepix_fly_client, "_on_received_ws_message") as mock_on_msg:
|
||||
timepix_fly_client._ws_send_and_receive()
|
||||
mock_on_msg.assert_called_once_with("init")
|
||||
|
||||
# Now simulate a TimeoutError from recv: should simply return and not call _on_received_ws_message
|
||||
mock_ws2 = mock.Mock()
|
||||
mock_ws2.recv.side_effect = TimeoutError
|
||||
timepix_fly_client.ws_client = mock_ws2
|
||||
with mock.patch.object(timepix_fly_client, "_on_received_ws_message") as mock_on_msg2:
|
||||
timepix_fly_client._ws_send_and_receive()
|
||||
mock_on_msg2.assert_not_called()
|
||||
|
||||
|
||||
def test_timepix_fly_client_on_message_received(timepix_fly_client):
|
||||
"""This tests the _on_received_ws_message method of timepix fly client."""
|
||||
with mock.patch.object(timepix_fly_client, "_run_status_callbacks") as mock_run_callbacks:
|
||||
timepix_fly_client._on_received_ws_message("init")
|
||||
assert timepix_fly_client._status == TimePixFlyStatus.INIT
|
||||
mock_run_callbacks.assert_called_once()
|
||||
|
||||
# invalid message should not change status or call callbacks
|
||||
prev_status = timepix_fly_client._status
|
||||
with mock.patch.object(timepix_fly_client, "_run_status_callbacks") as mock_run_callbacks2:
|
||||
timepix_fly_client._on_received_ws_message("invalid_status_string")
|
||||
# status stays as previous value and callbacks not called
|
||||
assert timepix_fly_client._status == prev_status
|
||||
mock_run_callbacks2.assert_not_called()
|
||||
|
||||
|
||||
def test_timepix_fly_client_on_status_callbacks(timepix_fly_client):
|
||||
"""This tests the _run_status_callbacks method of timepix fly client."""
|
||||
# Immediate run when current status already in success
|
||||
timepix_fly_client._status = TimePixFlyStatus.INIT
|
||||
status = StatusBase()
|
||||
timepix_fly_client.add_status_callback(
|
||||
status=status, success=[TimePixFlyStatus.INIT], error=[TimePixFlyStatus.EXCEPT], run=True
|
||||
)
|
||||
assert status.done is True and status.success is True
|
||||
|
||||
# Add callback (do not run immediately) and then trigger via _run_status_callbacks
|
||||
status2 = StatusBase()
|
||||
timepix_fly_client.add_status_callback(
|
||||
status=status2,
|
||||
success=[TimePixFlyStatus.CONFIG],
|
||||
error=[TimePixFlyStatus.EXCEPT],
|
||||
run=False,
|
||||
)
|
||||
# Set status to CONFIG and mark started True to check reset
|
||||
timepix_fly_client._status = TimePixFlyStatus.CONFIG
|
||||
timepix_fly_client._started = True
|
||||
timepix_fly_client._run_status_callbacks()
|
||||
assert status2.done is True and status2.success is True
|
||||
assert timepix_fly_client._started is False
|
||||
|
||||
# Error path: add with run True when status is EXCEPT
|
||||
timepix_fly_client._status = TimePixFlyStatus.EXCEPT
|
||||
status3 = StatusBase()
|
||||
with mock.patch.object(timepix_fly_client, "last_error") as mock_last_error:
|
||||
mock_err = mock.Mock()
|
||||
mock_err.message = "boom"
|
||||
mock_last_error.return_value = mock_err
|
||||
|
||||
timepix_fly_client.add_status_callback(
|
||||
status=status3,
|
||||
success=[TimePixFlyStatus.INIT],
|
||||
error=[TimePixFlyStatus.EXCEPT],
|
||||
run=True,
|
||||
)
|
||||
assert status3.done is True and status3.success is False
|
||||
|
||||
|
||||
def test_timepix_fly_client_shutdown(timepix_fly_client):
|
||||
"""This tests the shutdown method of timepix fly client."""
|
||||
mock_ws = mock.Mock()
|
||||
timepix_fly_client.ws_client = mock_ws
|
||||
|
||||
timepix_fly_client.shutdown()
|
||||
|
||||
mock_ws.close.assert_called_once()
|
||||
assert timepix_fly_client.ws_client is None
|
||||
assert timepix_fly_client._shutdown_event.is_set()
|
||||
|
||||
|
||||
def test_timepix_fly_client_start(timepix_fly_client):
|
||||
"""This tests the start method of timepix fly client."""
|
||||
# The client._get uses requests.get with f"http://{self.rest_url}/{get_cmd}".
|
||||
# We mock requests.get and verify it is called with the expected URL and timeout,
|
||||
# and that the _started flag is set.
|
||||
module_path = "superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_client"
|
||||
with mock.patch(f"{module_path}.requests.get") as mock_get:
|
||||
mock_resp = mock.Mock()
|
||||
mock_resp.raise_for_status = mock.Mock()
|
||||
mock_resp.text = ""
|
||||
mock_get.return_value = mock_resp
|
||||
|
||||
timepix_fly_client.start()
|
||||
|
||||
expected_url = f"http://{timepix_fly_client.rest_url}/?start=true"
|
||||
mock_get.assert_called_once_with(expected_url, timeout=timepix_fly_client._timeout)
|
||||
assert timepix_fly_client._started is True
|
||||
|
||||
|
||||
def test_timepix_fly_client_stop_collect(timepix_fly_client):
|
||||
"""This tests the stop_collect method of timepix fly client."""
|
||||
module_path = "superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_client"
|
||||
with mock.patch(f"{module_path}.requests.get") as mock_get:
|
||||
mock_resp = mock.Mock()
|
||||
mock_resp.raise_for_status = mock.Mock()
|
||||
mock_resp.text = ""
|
||||
mock_get.return_value = mock_resp
|
||||
|
||||
# ensure started is True and then call stop_collect
|
||||
timepix_fly_client._started = True
|
||||
timepix_fly_client.stop_collect()
|
||||
|
||||
expected_url = f"http://{timepix_fly_client.rest_url}/?stop_collect=true"
|
||||
mock_get.assert_called_once_with(expected_url, timeout=timepix_fly_client._timeout)
|
||||
assert timepix_fly_client._started is False
|
||||
|
||||
|
||||
def test_timepix_fly_client_state(timepix_fly_client):
|
||||
"""This tests the state method of timepix fly client."""
|
||||
module_path = "superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_client"
|
||||
with mock.patch(f"{module_path}.requests.get") as mock_get:
|
||||
mock_resp = mock.Mock()
|
||||
mock_resp.raise_for_status = mock.Mock()
|
||||
# Return a JSON payload compatible with ProgramState
|
||||
mock_resp.json.return_value = {"type": "ProgramState", "state": "init"}
|
||||
mock_get.return_value = mock_resp
|
||||
|
||||
program_state = timepix_fly_client.state()
|
||||
|
||||
expected_url = f"http://{timepix_fly_client.rest_url}/state"
|
||||
mock_get.assert_called_once_with(expected_url, timeout=timepix_fly_client._timeout)
|
||||
# ProgramState.model defines 'state' as a string literal; ensure we parsed it
|
||||
assert hasattr(program_state, "state")
|
||||
assert program_state.state == "init"
|
||||
|
||||
|
||||
def test_timepix_fly_client_set_pixel_map(timepix_fly_client):
|
||||
"""This tests the set_pixel_map/_put path by mocking requests.put and checking payload."""
|
||||
module_path = "superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_client"
|
||||
with mock.patch(f"{module_path}.requests.put") as mock_put:
|
||||
mock_resp = mock.Mock()
|
||||
mock_resp.raise_for_status = mock.Mock()
|
||||
mock_put.return_value = mock_resp
|
||||
|
||||
# Minimal valid PixelMap dict (type is optional; model supplies default)
|
||||
pixel_map = {"chips": [[{"i": 0, "p": 1, "f": 2}]]}
|
||||
|
||||
timepix_fly_client.set_pixel_map(pixel_map)
|
||||
|
||||
expected_url = f"http://{timepix_fly_client.rest_url}/pixel-map"
|
||||
# Verify requests.put called with expected url, json and timeout
|
||||
mock_put.assert_called_once()
|
||||
_, kwargs = mock_put.call_args
|
||||
assert kwargs.get("timeout") == timepix_fly_client._timeout
|
||||
assert kwargs.get("json") is not None
|
||||
assert kwargs.get("json").get("chips") == pixel_map["chips"]
|
||||
Reference in New Issue
Block a user