mirror of
https://github.com/slsdetectorgroup/aare.git
synced 2026-11-09 11:43:54 +01:00
Merge branch 'main' into testing/cuda_cf_api
Three files conflicted: - ClusterFinderMT::collect(): both sides rewrote it. Keep this branch's version (f4bbbc9), which reads the stop flags before sweeping, over main's (afce91b), which still samples queue emptiness mid-condition. The staged splice had also lost a brace, swallowing popFront(). - python/aare/ClusterFinder.py: restore the CUDA API deleted by the resolution (_cuda_available, _CUDA_ALGORITHMS, ClusterFinderCUDA, find_cluster_views_batched_iter, ClusterFinderCUDAGraph), and main's min_pedestal_samples forwarding, dropped as an adjacent hunk. - bind_ClusterVector.hpp: keep py::module_local() (133cedf) alongside main's new docstrings; without it _aare and _aare_cuda cannot both register Cluster/ClusterVector under pybind11 3. Does not build on its own: ClusterFinderFrozen and the CUDA drivers still call Pedestal APIs main removed. Fixed in the follow-up.
This commit is contained in:
kferjaoui committed
commit
7a8f918a56
91 files changed
+5740
-1303
No files matched your search
@@ -3,7 +3,7 @@ import pytest
|
||||
import numpy as np
|
||||
|
||||
from aare import _aare #import the C++ module
|
||||
from aare import corner
|
||||
from aare import ROI, corner
|
||||
from conftest import test_data_path
|
||||
|
||||
|
||||
@@ -91,6 +91,22 @@ def test_calculate_eta():
|
||||
assert eta2.sum == 4
|
||||
|
||||
|
||||
def test_cluster_representations():
|
||||
"""Test Cluster representation"""
|
||||
|
||||
cluster = _aare.Cluster2x2i(5, 6, np.array([1, 2, 3, 4], dtype=np.int32))
|
||||
assert repr(cluster) == "Cluster2x2i(x=5, y=6, data=[[1, 2], [3, 4]])"
|
||||
|
||||
def test_eta_representations():
|
||||
"""Test Eta representation"""
|
||||
eta = _aare.Etai()
|
||||
eta.x = 0.25
|
||||
eta.y = 0.75
|
||||
eta.c = corner.cBottomRight
|
||||
eta.sum = 42
|
||||
assert repr(eta) == "Etai(x=0.25, y=0.75, c=BottomRight, sum=42)"
|
||||
|
||||
|
||||
def test_max_sum():
|
||||
"""Max 2x2 Sum"""
|
||||
cluster = _aare.Cluster3x3i(5,5,np.array([1, 1, 1, 2, 3, 1, 2, 2, 1], dtype=np.int32))
|
||||
@@ -146,4 +162,3 @@ def test_3x3_reduction():
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -6,10 +6,279 @@ import boost_histogram as bh
|
||||
import time
|
||||
from pathlib import Path
|
||||
import pickle
|
||||
import struct
|
||||
import gc
|
||||
import weakref
|
||||
|
||||
from aare import ClusterFile
|
||||
from aare import ClusterFile, ROI
|
||||
from conftest import test_data_path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def iterator_file(tmp_path):
|
||||
fname = tmp_path / "iteration.clust"
|
||||
records = [
|
||||
(-11, []),
|
||||
(42, [(5, 6, 10), (7, 8, 20)]),
|
||||
(103, []),
|
||||
(104, [(5, 6, 30), (7, 8, 40)]),
|
||||
(105, []),
|
||||
]
|
||||
fname.write_bytes(b"".join(
|
||||
struct.pack("@iI", number, len(clusters))
|
||||
+ b"".join(struct.pack("@HH9i", x, y, *([value] * 9))
|
||||
for x, y, value in clusters)
|
||||
for number, clusters in records
|
||||
))
|
||||
return fname
|
||||
|
||||
|
||||
def test_frames_preserve_empty_frames_and_frame_numbers(iterator_file):
|
||||
with ClusterFile(iterator_file) as reader:
|
||||
frames = reader.frames()
|
||||
assert iter(frames) is frames
|
||||
assert reader.tell() == 0
|
||||
results = list(frames)
|
||||
assert [frame.frame_number for frame in results] == [-11, 42, 103, 104, 105]
|
||||
assert [frame.size for frame in results] == [0, 2, 0, 2, 0]
|
||||
for _ in range(2):
|
||||
with pytest.raises(StopIteration):
|
||||
next(frames)
|
||||
assert list(reader.frames()) == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize("chunk_size, sizes", [(1, [1, 1, 1, 1]), (2, [2, 2]),
|
||||
(3, [3, 1]), (10, [4])])
|
||||
@pytest.mark.parametrize("explicit", [False, True])
|
||||
def test_chunks_cross_frames(iterator_file, chunk_size, sizes, explicit):
|
||||
with ClusterFile(iterator_file, chunk_size=chunk_size) as reader:
|
||||
chunks = reader.chunks(chunk_size=chunk_size) if explicit else reader.chunks()
|
||||
assert iter(chunks) is chunks
|
||||
assert reader.tell() == 0
|
||||
results = list(chunks)
|
||||
assert [chunk.size for chunk in results] == sizes
|
||||
data = np.concatenate([np.asarray(chunk) for chunk in results])
|
||||
np.testing.assert_array_equal(data["x"], [5, 7, 5, 7])
|
||||
np.testing.assert_array_equal(data["data"][:, 0, 0], [10, 20, 30, 40])
|
||||
for _ in range(2):
|
||||
with pytest.raises(StopIteration):
|
||||
next(chunks)
|
||||
|
||||
|
||||
def test_chunk_override_preserves_default_iteration(iterator_file):
|
||||
with ClusterFile(iterator_file, chunk_size=2) as reader:
|
||||
assert iter(reader) is reader
|
||||
assert next(reader.chunks(1)).size == 1
|
||||
assert [chunk.size for chunk in reader] == [2, 1]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("use_roi", [False, True])
|
||||
@pytest.mark.parametrize("use_noise", [False, True])
|
||||
@pytest.mark.parametrize("method", ["frames", "chunks"])
|
||||
def test_iterators_apply_filters_and_gain(iterator_file, use_roi, use_noise, method):
|
||||
with ClusterFile(iterator_file, chunk_size=3) as reader:
|
||||
if use_roi:
|
||||
reader.set_roi(ROI(0, 6, 0, 12))
|
||||
if use_noise:
|
||||
reader.set_noise_map(np.full((12, 12), 15, dtype=np.int32))
|
||||
reader.set_gain_map(np.full((12, 12), 2.0))
|
||||
results = list(getattr(reader, method)())
|
||||
if method == "frames":
|
||||
assert [frame.frame_number for frame in results] == [-11, 42, 103, 104, 105]
|
||||
expected = [value / 2 for x, value in [(5, 10), (7, 20), (5, 30), (7, 40)]
|
||||
if (not use_roi or x == 5) and (not use_noise or value > 15)]
|
||||
data = np.concatenate([np.asarray(result) for result in results])
|
||||
np.testing.assert_array_equal(data["data"][:, 0, 0], expected)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("method", ["frames", "chunks"])
|
||||
def test_iterators_handle_all_rejected_clusters(iterator_file, method):
|
||||
with ClusterFile(iterator_file) as reader:
|
||||
reader.set_roi(ROI(0, 1, 0, 1))
|
||||
results = list(getattr(reader, method)())
|
||||
assert len(results) == (5 if method == "frames" else 0)
|
||||
assert all(result.size == 0 for result in results)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("method", ["frames", "chunks"])
|
||||
def test_iterator_keeps_file_alive(iterator_file, method):
|
||||
reader = ClusterFile(iterator_file)
|
||||
owner = weakref.ref(reader)
|
||||
iterator = getattr(reader, method)()
|
||||
del reader
|
||||
gc.collect()
|
||||
assert owner() is not None
|
||||
results = list(iterator)
|
||||
assert sum(result.size for result in results) == 4
|
||||
del iterator
|
||||
gc.collect()
|
||||
assert owner() is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("method", ["frames", "chunks"])
|
||||
def test_retained_results_and_numpy_views_own_storage(iterator_file, method):
|
||||
with ClusterFile(iterator_file, chunk_size=2) as reader:
|
||||
reader.read_frame() # Skip the initial empty frame.
|
||||
iterator = getattr(reader, method)()
|
||||
first = next(iterator)
|
||||
array = np.asarray(first)
|
||||
expected = array.copy()
|
||||
results = list(iterator)
|
||||
np.testing.assert_array_equal(np.asarray(first), expected)
|
||||
del first, results, iterator, reader
|
||||
gc.collect()
|
||||
np.testing.assert_array_equal(array, expected)
|
||||
|
||||
|
||||
def test_frame_iteration_resumes_without_reading_ahead(iterator_file):
|
||||
with ClusterFile(iterator_file) as reader:
|
||||
reader.read_frame()
|
||||
for frame in reader.frames():
|
||||
assert frame.frame_number == 42
|
||||
break
|
||||
assert reader.read_frame().frame_number == 103
|
||||
assert [frame.frame_number for frame in reader.frames()] == [104, 105]
|
||||
|
||||
|
||||
def test_partial_chunk_must_be_completed_before_frame_iteration(iterator_file):
|
||||
with ClusterFile(iterator_file) as reader:
|
||||
for chunk in reader.chunks(1):
|
||||
assert chunk.size == 1
|
||||
break
|
||||
with pytest.raises(RuntimeError, match="clusters left"):
|
||||
next(reader.frames())
|
||||
assert reader.read_clusters(1).size == 1
|
||||
assert [frame.frame_number for frame in reader.frames()] == [103, 104, 105]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("method", ["frames", "chunks"])
|
||||
def test_empty_iterators(tmp_path, method):
|
||||
fname = tmp_path / "empty.clust"
|
||||
fname.touch()
|
||||
with ClusterFile(fname) as reader:
|
||||
iterator = getattr(reader, method)()
|
||||
for _ in range(2):
|
||||
with pytest.raises(StopIteration):
|
||||
next(iterator)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("method", ["frames", "chunks"])
|
||||
@pytest.mark.parametrize("started", [False, True])
|
||||
def test_iterators_reject_closed_files(iterator_file, method, started):
|
||||
reader = ClusterFile(iterator_file, chunk_size=1)
|
||||
iterator = getattr(reader, method)()
|
||||
if started:
|
||||
next(iterator)
|
||||
reader.close()
|
||||
with pytest.raises(RuntimeError, match="not opened for reading"):
|
||||
next(iterator)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("method", ["frames", "chunks"])
|
||||
@pytest.mark.parametrize("mode", ["w", "a"])
|
||||
def test_iterators_reject_writing_modes(tmp_path, method, mode):
|
||||
with ClusterFile(tmp_path / "output.clust", mode=mode) as reader:
|
||||
with pytest.raises(RuntimeError, match="not opened for reading"):
|
||||
next(getattr(reader, method)())
|
||||
|
||||
|
||||
def test_chunk_sizes_must_be_positive(iterator_file):
|
||||
with ClusterFile(iterator_file, chunk_size=0) as reader:
|
||||
with pytest.raises(ValueError, match="greater than zero"):
|
||||
reader.chunks()
|
||||
with pytest.raises(ValueError, match="greater than zero"):
|
||||
reader.chunks(0)
|
||||
with pytest.raises(ValueError, match="greater than zero"):
|
||||
next(reader)
|
||||
with pytest.raises(TypeError):
|
||||
reader.chunks(-1)
|
||||
assert reader.tell() == 0
|
||||
assert reader.read_clusters(0).size == 0
|
||||
assert len(list(reader.frames())) == 5
|
||||
|
||||
|
||||
@pytest.mark.parametrize("method", ["frames", "chunks"])
|
||||
def test_iterators_defer_read_errors_until_next_result(tmp_path, method):
|
||||
fname = tmp_path / "truncated.clust"
|
||||
cluster = struct.pack("@HH9i", 5, 6, *range(9))
|
||||
frame = struct.pack("@iI", 42, 1) + cluster
|
||||
fname.write_bytes(frame + frame[:-1])
|
||||
with ClusterFile(fname, chunk_size=1) as reader:
|
||||
iterator = getattr(reader, method)()
|
||||
assert next(iterator).size == 1
|
||||
assert reader.tell() == len(frame)
|
||||
with pytest.raises(RuntimeError):
|
||||
next(iterator)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"shape, dtype",
|
||||
[(shape, dtype)
|
||||
for shape in [(2, 2), (3, 3), (5, 5), (7, 7), (9, 9)]
|
||||
for dtype in [np.int32, np.float32, np.float64]]
|
||||
+ [((3, 3), np.int16)],
|
||||
)
|
||||
def test_iterators_are_bound_for_each_cluster_type(tmp_path, shape, dtype):
|
||||
fname = tmp_path / "typed_frames.clust"
|
||||
values = list(range(shape[0] * shape[1]))
|
||||
record = struct.pack("@HH" + np.dtype(dtype).char * len(values), 5, 6, *values)
|
||||
fname.write_bytes((struct.pack("@iI", -42, 1) + record) * 3)
|
||||
with ClusterFile(fname, cluster_size=shape, dtype=dtype) as reader:
|
||||
frame = next(reader.frames())
|
||||
assert frame.frame_number == -42
|
||||
np.testing.assert_array_equal(np.asarray(frame)["data"].reshape(-1), values)
|
||||
chunks = list(reader.chunks())
|
||||
assert len(chunks) == 1
|
||||
assert chunks[0].size == 2
|
||||
np.testing.assert_array_equal(np.asarray(chunks[0])["data"].reshape(-1), values * 2)
|
||||
|
||||
|
||||
def test_read_frame_returns_none_at_eof(tmp_path):
|
||||
fname = tmp_path / "empty.clust"
|
||||
fname.touch()
|
||||
|
||||
with ClusterFile(fname) as f:
|
||||
assert f.read_frame() is None
|
||||
|
||||
|
||||
def test_read_frame_raises_for_malformed_file(tmp_path):
|
||||
fname = tmp_path / "malformed.clust"
|
||||
fname.write_bytes(b"\x00")
|
||||
|
||||
with ClusterFile(fname) as f, pytest.raises(RuntimeError):
|
||||
f.read_frame()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("use_roi", [False, True])
|
||||
@pytest.mark.parametrize("method", ["read_clusters", "default", "frames", "chunks"])
|
||||
@pytest.mark.parametrize(
|
||||
"size",
|
||||
[
|
||||
pytest.param(3, id="partial-frame-number"),
|
||||
pytest.param(4, id="missing-cluster-count"),
|
||||
pytest.param(7, id="partial-cluster-count"),
|
||||
pytest.param(48, id="missing-cluster-record"),
|
||||
pytest.param(87, id="partial-cluster-record"),
|
||||
],
|
||||
)
|
||||
def test_chunk_reads_reject_incomplete_frames(tmp_path, use_roi, method, size):
|
||||
fname = tmp_path / "incomplete.clust"
|
||||
cluster = struct.pack("@HH9i", 5, 6, *range(9))
|
||||
frame = struct.pack("@iI", 42, 2) + cluster * 2
|
||||
fname.write_bytes(frame[:size])
|
||||
|
||||
with ClusterFile(fname) as reader:
|
||||
if use_roi:
|
||||
reader.set_roi(ROI(0, 10, 0, 10))
|
||||
with pytest.raises(RuntimeError):
|
||||
if method == "default":
|
||||
next(reader)
|
||||
elif method == "read_clusters":
|
||||
reader.read_clusters(10)
|
||||
else:
|
||||
next(getattr(reader, method)())
|
||||
|
||||
|
||||
@pytest.mark.withdata
|
||||
def test_cluster_file(test_data_path):
|
||||
"""Test ClusterFile"""
|
||||
|
||||
@@ -83,12 +83,15 @@ def test_make_a_hitmap_from_cluster_vector():
|
||||
|
||||
def test_2x2_reduction():
|
||||
cv = ClusterVector((3,3))
|
||||
cv.frame_number = -135
|
||||
|
||||
cv.push_back(_aare.Cluster3x3i(5, 5, np.array([1, 1, 1, 2, 3, 1, 2, 2, 1], dtype=np.int32)))
|
||||
cv.push_back(_aare.Cluster3x3i(5, 5, np.array([2, 2, 1, 2, 3, 1, 1, 1, 1], dtype=np.int32)))
|
||||
|
||||
reduced_cv = np.array(_aare.reduce_to_2x2(cv), copy=False)
|
||||
reduced = _aare.reduce_to_2x2(cv)
|
||||
reduced_cv = np.array(reduced, copy=False)
|
||||
|
||||
assert reduced.frame_number == cv.frame_number
|
||||
assert reduced_cv.size == 2
|
||||
assert reduced_cv[0]["x"] == 5
|
||||
assert reduced_cv[0]["y"] == 5
|
||||
@@ -100,14 +103,17 @@ def test_2x2_reduction():
|
||||
|
||||
def test_3x3_reduction():
|
||||
cv = _aare.ClusterVector_Cluster5x5d()
|
||||
cv.frame_number = 246
|
||||
|
||||
cv.push_back(_aare.Cluster5x5d(5,5,np.array([1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 2.0, 1.0, 1.0, 1.0, 2.0, 2.0, 3.0,
|
||||
1.0, 1.0, 1.0, 2.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0], dtype=np.double)))
|
||||
cv.push_back(_aare.Cluster5x5d(5,5,np.array([1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 2.0, 1.0, 1.0, 1.0, 2.0, 2.0, 3.0,
|
||||
1.0, 1.0, 1.0, 2.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0], dtype=np.double)))
|
||||
|
||||
reduced_cv = np.array(_aare.reduce_to_3x3(cv), copy=False)
|
||||
reduced = _aare.reduce_to_3x3(cv)
|
||||
reduced_cv = np.array(reduced, copy=False)
|
||||
|
||||
assert reduced.frame_number == cv.frame_number
|
||||
assert reduced_cv.size == 2
|
||||
assert reduced_cv[0]["x"] == 5
|
||||
assert reduced_cv[0]["y"] == 5
|
||||
@@ -130,3 +136,26 @@ def test_masking():
|
||||
assert cv_masked_array[0]["x"] == 1
|
||||
assert cv_masked_array[0]["y"] == 2
|
||||
assert (cv_masked_array[0]["data"] == np.ones((3,3),dtype=np.int32)).all()
|
||||
|
||||
|
||||
def test_masking_requires_c_contiguous_array():
|
||||
cv = _aare.ClusterVector_Cluster3x3i()
|
||||
cv.push_back(_aare.Cluster3x3i(1, 2, np.ones(9, dtype=np.int32)))
|
||||
cv.push_back(_aare.Cluster3x3i(3, 4, np.ones(9, dtype=np.int32)))
|
||||
|
||||
mask = np.array([True, False, True, False], dtype=bool)[::2]
|
||||
assert not mask.flags.c_contiguous
|
||||
|
||||
with pytest.raises(TypeError):
|
||||
cv(mask)
|
||||
|
||||
|
||||
def test_masking_requires_one_dimension():
|
||||
cv = _aare.ClusterVector_Cluster3x3i()
|
||||
cv.push_back(_aare.Cluster3x3i(1, 2, np.ones(9, dtype=np.int32)))
|
||||
cv.push_back(_aare.Cluster3x3i(3, 4, np.ones(9, dtype=np.int32)))
|
||||
|
||||
mask = np.array([[True, False]], dtype=bool)
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
cv(mask)
|
||||
@@ -0,0 +1,58 @@
|
||||
|
||||
|
||||
import pytest
|
||||
import numpy as np
|
||||
from aare import CtbRawFile
|
||||
|
||||
|
||||
@pytest.mark.withdata
|
||||
def test_basic_properites(test_data_path):
|
||||
f = CtbRawFile(test_data_path/'raw/ctb/run_master_0.json')
|
||||
assert f.total_frames == 47
|
||||
assert f.image_size_in_bytes == 640
|
||||
assert f.master.analog_samples == 10
|
||||
assert f.master.digital_samples is None
|
||||
assert f.master.transceiver_samples is None
|
||||
|
||||
|
||||
@pytest.mark.withdata
|
||||
def test_seek_and_read(test_data_path):
|
||||
f = CtbRawFile(test_data_path/'raw/ctb/run_master_0.json')
|
||||
f.seek(0)
|
||||
header, raw_data = f.read_frame()
|
||||
assert header['frameNumber'] == 48 #we know that the first frame in this file is 48
|
||||
|
||||
#Seek a multiple of frames in file
|
||||
f.seek(9)
|
||||
header, raw_data = f.read_frame()
|
||||
assert header['frameNumber'] == 57
|
||||
|
||||
@pytest.mark.withdata
|
||||
def test_read_all_frames_direct_and_iterate(test_data_path):
|
||||
"""
|
||||
We want to make sure that iterating and reading give the same result.
|
||||
"""
|
||||
f0 = CtbRawFile(test_data_path/'raw/ctb/run_master_0.json')
|
||||
f1 = CtbRawFile(test_data_path/'raw/ctb/run_master_0.json')
|
||||
|
||||
for h0, d0 in f0:
|
||||
h1, d1 = f1.read_frame()
|
||||
assert h0 == h1
|
||||
assert np.all(d0 == d1)
|
||||
|
||||
@pytest.mark.withdata
|
||||
def test_read_all_frames_with_index_and_iterate(test_data_path):
|
||||
"""
|
||||
We want to make sure that iterating and reading give the same result.
|
||||
"""
|
||||
f0 = CtbRawFile(test_data_path/'raw/ctb/run_master_0.json')
|
||||
f1 = CtbRawFile(test_data_path/'raw/ctb/run_master_0.json')
|
||||
|
||||
frames_read = 0
|
||||
for i,(h0, d0) in enumerate(f0):
|
||||
h1, d1 = f1.read_frame(i)
|
||||
assert h0 == h1
|
||||
assert np.all(d0 == d1)
|
||||
frames_read += 1
|
||||
|
||||
assert frames_read == f0.total_frames
|
||||
@@ -143,3 +143,17 @@ def test_fast_pedestal_rejects_wrong_shape():
|
||||
|
||||
with pytest.raises(RuntimeError, match="shape"):
|
||||
pedestal.add_init_frame(np.zeros((2, 2), dtype=np.uint16))
|
||||
|
||||
|
||||
@pytest.mark.parametrize("dtype", [np.float64, np.float32, np.int16])
|
||||
def test_fast_pedestal_std_uses_double_variance(dtype):
|
||||
pedestal = FastPedestal(1, 1, n_samples=2, dtype=dtype)
|
||||
pedestal.add_init_frame(np.array([[0]], dtype=np.uint16))
|
||||
pedestal.add_init_frame(np.array([[1000]], dtype=np.uint16))
|
||||
|
||||
assert pedestal.std().dtype == dtype
|
||||
np.testing.assert_array_equal(pedestal.std(), [[500]])
|
||||
|
||||
pedestal.push_ema(np.array([[1000]], dtype=np.uint16))
|
||||
expected = np.array([[np.sqrt(187500.0)]], dtype=dtype)
|
||||
np.testing.assert_array_equal(pedestal.std(), expected)
|
||||
@@ -1,7 +1,106 @@
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from aare import Pedestal_d, Pedestal_f
|
||||
from aare import Pedestal, Pedestal_d, Pedestal_f, Pedestal_i16
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("dtype", "pedestal_type"),
|
||||
[(np.float64, Pedestal_d), (np.float32, Pedestal_f), (np.int16, Pedestal_i16)],
|
||||
)
|
||||
def test_pedestal_factory(dtype, pedestal_type):
|
||||
pedestal = Pedestal(rows=2, cols=3, n_samples=4, dtype=dtype)
|
||||
|
||||
assert isinstance(pedestal, pedestal_type)
|
||||
assert pedestal.rows == 2
|
||||
assert pedestal.cols == 3
|
||||
assert pedestal.n_samples == 4
|
||||
|
||||
|
||||
def test_pedestal_factory_defaults_to_double():
|
||||
pedestal = Pedestal(2, 3)
|
||||
|
||||
assert isinstance(pedestal, Pedestal_d)
|
||||
assert pedestal.n_samples == 1000
|
||||
|
||||
|
||||
def test_pedestal_factory_rejects_unbound_dtype():
|
||||
with pytest.raises(ValueError, match="Unsupported dtype for Pedestal"):
|
||||
Pedestal(2, 3, dtype=np.int32)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("dtype", [np.float64, np.float32, np.int16])
|
||||
@pytest.mark.parametrize("name", ["rows", "cols", "n_samples"])
|
||||
def test_pedestal_factory_rejects_negative_parameters(dtype, name):
|
||||
parameters = dict(rows=1, cols=1, n_samples=10)
|
||||
parameters[name] = -1
|
||||
with pytest.raises(TypeError):
|
||||
Pedestal(**parameters, dtype=dtype)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("pedestal_type", [Pedestal_d, Pedestal_f, Pedestal_i16])
|
||||
@pytest.mark.parametrize(
|
||||
"args", [(-1, 1), (1, -1), (-1, 1, 10), (1, -1, 10), (1, 1, -1)]
|
||||
)
|
||||
def test_pedestal_constructor_rejects_negative_parameters(pedestal_type, args):
|
||||
with pytest.raises(TypeError):
|
||||
pedestal_type(*args)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("dtype", [np.float64, np.float32, np.int16])
|
||||
def test_pedestal_std_stays_finite_after_settling(dtype):
|
||||
pedestal = Pedestal(1, 1, n_samples=10, dtype=dtype)
|
||||
pedestal.push(np.array([[16382]], dtype=np.uint16))
|
||||
frame = np.array([[16383]], dtype=np.uint16)
|
||||
for _ in range(201):
|
||||
pedestal.push(frame)
|
||||
|
||||
noise = pedestal.std().item()
|
||||
assert np.isfinite(noise)
|
||||
assert 0 <= noise < 1e-3
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("pedestal_type", "expected_dtype"),
|
||||
[(Pedestal_d, np.float64), (Pedestal_f, np.float32), (Pedestal_i16, np.int16)],
|
||||
)
|
||||
def test_double_precision_moments(pedestal_type, expected_dtype):
|
||||
pedestal = pedestal_type(1, 1, 2)
|
||||
for value in [30000, 30003, 30002]:
|
||||
pedestal.push(np.array([[value]], dtype=np.uint16))
|
||||
|
||||
assert pedestal.mean().dtype == expected_dtype
|
||||
assert pedestal.std().dtype == expected_dtype
|
||||
np.testing.assert_array_equal(
|
||||
pedestal.mean(), np.array([[30001.75]], dtype=expected_dtype)
|
||||
)
|
||||
np.testing.assert_allclose(
|
||||
pedestal.std(),
|
||||
np.array([[np.sqrt(1.1875)]], dtype=expected_dtype),
|
||||
rtol=1e-6,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("dtype", [np.float64, np.float32, np.int16])
|
||||
@pytest.mark.parametrize("method", ["push", "push_with_threshold"])
|
||||
@pytest.mark.parametrize("shape", [(2, 2), (3, 2)])
|
||||
def test_pedestal_rejects_mismatched_frame_shapes(dtype, method, shape):
|
||||
pedestal = Pedestal(2, 3, dtype=dtype)
|
||||
initial = np.full((2, 3), 7, dtype=np.uint16)
|
||||
pedestal.push(initial)
|
||||
frame = np.zeros(shape, dtype=np.uint16)
|
||||
args = (
|
||||
(np.full((2, 3), 10, dtype=dtype),)
|
||||
if method == "push_with_threshold"
|
||||
else ()
|
||||
)
|
||||
|
||||
with pytest.raises(
|
||||
RuntimeError, match="Frame shape does not match pedestal shape"
|
||||
):
|
||||
getattr(pedestal, method)(frame, *args)
|
||||
|
||||
np.testing.assert_array_equal(pedestal.mean(), initial)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
@@ -38,3 +137,68 @@ def test_pedestal_exposes_mean_as_read_only_buffer():
|
||||
np.testing.assert_array_equal(mean, pedestal.view())
|
||||
assert np.shares_memory(mean, pedestal.view())
|
||||
assert not mean.flags.writeable
|
||||
|
||||
|
||||
@pytest.mark.parametrize("dtype", [np.float64, np.float32, np.int16])
|
||||
@pytest.mark.parametrize("method", ["push", "push_with_threshold"])
|
||||
def test_pedestal_push_accepts_contiguous_frames(dtype, method):
|
||||
pedestal = Pedestal(2, 3, dtype=dtype)
|
||||
frame = np.arange(6, dtype=np.uint16).reshape(2, 3)
|
||||
args = (
|
||||
(np.full((2, 3), 10, dtype=dtype),)
|
||||
if method == "push_with_threshold"
|
||||
else ()
|
||||
)
|
||||
|
||||
getattr(pedestal, method)(frame, *args)
|
||||
|
||||
np.testing.assert_array_equal(pedestal.mean(), frame)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("dtype", [np.float64, np.float32, np.int16])
|
||||
@pytest.mark.parametrize("input_name", ["push_frame", "threshold_frame", "threshold"])
|
||||
@pytest.mark.parametrize("layout", ["transpose", "slice", "reverse"])
|
||||
def test_pedestal_rejects_noncontiguous_inputs(dtype, input_name, layout):
|
||||
pedestal = Pedestal(2, 3, dtype=dtype)
|
||||
frame = np.zeros((2, 3), dtype=np.uint16)
|
||||
threshold = np.full((2, 3), 10, dtype=dtype)
|
||||
input_dtype = dtype if input_name == "threshold" else np.uint16
|
||||
if layout == "transpose":
|
||||
invalid = np.ones((3, 2), dtype=input_dtype).T
|
||||
elif layout == "slice":
|
||||
invalid = np.ones((2, 6), dtype=input_dtype)[:, ::2]
|
||||
else:
|
||||
invalid = np.ones((4, 6), dtype=input_dtype)[:2, :3][:, ::-1]
|
||||
assert not invalid.flags.c_contiguous
|
||||
|
||||
with pytest.raises(TypeError):
|
||||
if input_name == "push_frame":
|
||||
pedestal.push(invalid)
|
||||
elif input_name == "threshold_frame":
|
||||
pedestal.push_with_threshold(invalid, threshold)
|
||||
else:
|
||||
pedestal.push_with_threshold(frame, invalid)
|
||||
|
||||
np.testing.assert_array_equal(pedestal.mean(), np.zeros((2, 3)))
|
||||
|
||||
|
||||
@pytest.mark.parametrize("dtype", [np.float64, np.float32, np.int16])
|
||||
@pytest.mark.parametrize("input_name", ["push_frame", "threshold_frame", "threshold"])
|
||||
@pytest.mark.parametrize("shape", [(), (6,), (2, 3, 2)])
|
||||
def test_pedestal_rejects_inputs_with_wrong_ndim(dtype, input_name, shape):
|
||||
pedestal = Pedestal(2, 3, dtype=dtype)
|
||||
frame = np.ones((2, 3), dtype=np.uint16)
|
||||
threshold = np.full((2, 3), 10, dtype=dtype)
|
||||
input_dtype = dtype if input_name == "threshold" else np.uint16
|
||||
invalid = np.ones(shape, dtype=input_dtype)
|
||||
name = "Threshold" if input_name == "threshold" else "Frame"
|
||||
|
||||
with pytest.raises(ValueError, match=f"{name} must be 2-dimensional"):
|
||||
if input_name == "push_frame":
|
||||
pedestal.push(invalid)
|
||||
elif input_name == "threshold_frame":
|
||||
pedestal.push_with_threshold(invalid, threshold)
|
||||
else:
|
||||
pedestal.push_with_threshold(frame, invalid)
|
||||
|
||||
np.testing.assert_array_equal(pedestal.mean(), np.zeros((2, 3)))
|
||||
@@ -0,0 +1,21 @@
|
||||
import pytest
|
||||
|
||||
import numpy as np
|
||||
|
||||
from aare import ROI
|
||||
|
||||
|
||||
def test_roi_slice():
|
||||
"""test the ROI slice method"""
|
||||
roi = ROI(0, 4, 2, 4)
|
||||
|
||||
array = np.array([[1, 2, 3, 4],
|
||||
[5, 6, 7, 8],
|
||||
[9, 10, 11, 12],
|
||||
[13, 14, 15, 16]])
|
||||
|
||||
sliced_array = array[roi.slice()]
|
||||
|
||||
assert sliced_array.shape == (2, 4)
|
||||
assert np.array_equal(sliced_array, np.array([[9, 10, 11, 12],
|
||||
[13, 14, 15, 16]]))
|
||||
@@ -1,8 +1,299 @@
|
||||
# SPDX-License-Identifier: MPL-2.0
|
||||
import pytest
|
||||
from aare import RawFile
|
||||
import json
|
||||
from aare import File, RawFile, RawSubFile, DetectorType, ROI, UDPPortPosition
|
||||
import numpy as np
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def small_raw_file(tmp_path):
|
||||
master_path = tmp_path / "run_master_0.json"
|
||||
master_path.write_text(json.dumps({
|
||||
"Version": 7.2,
|
||||
"Detector Type": "Jungfrau",
|
||||
"Timing Mode": "auto",
|
||||
"Geometry": {"x": 1, "y": 1},
|
||||
"Image Size in bytes": 12,
|
||||
"Pixels": {"x": 3, "y": 2},
|
||||
"Max Frames Per File": 1,
|
||||
"Total Frames": 2,
|
||||
"Frames in File": 2,
|
||||
"Frame Padding": 1,
|
||||
"Frame Discard Policy": "nodiscard",
|
||||
}))
|
||||
for index in range(2):
|
||||
data_path = tmp_path / f"run_d0_f{index}_0.raw"
|
||||
# A detector header followed by six uint16 pixels.
|
||||
data_path.write_bytes(bytes(112) + np.arange(6, dtype=np.uint16).tobytes())
|
||||
return master_path
|
||||
|
||||
|
||||
@pytest.mark.parametrize("reader_type", [RawFile, File])
|
||||
@pytest.mark.parametrize("legacy_master", [False, True])
|
||||
@pytest.mark.parametrize("padding, policy, supported", [
|
||||
(0, "nodiscard", False),
|
||||
(0, "discard", False),
|
||||
(0, "discardpartial", True),
|
||||
(1, "nodiscard", True),
|
||||
(1, "discard", True),
|
||||
(1, "discardpartial", True),
|
||||
])
|
||||
def test_raw_frame_policy(small_raw_file, reader_type, legacy_master,
|
||||
padding, policy, supported):
|
||||
master_path = small_raw_file
|
||||
if legacy_master:
|
||||
master_path = master_path.with_suffix(".raw")
|
||||
master_path.write_text(
|
||||
"Version : 6.4\n"
|
||||
"Detector Type : Jungfrau\n"
|
||||
"Timing Mode : auto\n"
|
||||
"Geometry : [1, 1]\n"
|
||||
"Image Size : 12\n"
|
||||
"Pixels : [3, 2]\n"
|
||||
"Dynamic Range : 16\n"
|
||||
"Max Frames Per File : 1\n"
|
||||
"Total Frames : 2\n"
|
||||
"Frames in File : 2\n"
|
||||
f"Frame Padding : {padding}\n"
|
||||
f"Frame Discard Policy : {policy}\n"
|
||||
)
|
||||
else:
|
||||
metadata = json.loads(master_path.read_text())
|
||||
metadata["Frame Padding"] = padding
|
||||
metadata["Frame Discard Policy"] = policy
|
||||
master_path.write_text(json.dumps(metadata))
|
||||
|
||||
if supported:
|
||||
reader = reader_type(master_path)
|
||||
assert reader.total_frames == 2
|
||||
frame = reader.read_frame()
|
||||
if reader_type is RawFile:
|
||||
_, frame = frame
|
||||
np.testing.assert_array_equal(frame, np.arange(6).reshape(2, 3))
|
||||
else:
|
||||
message = "requires frame padding or discardpartial"
|
||||
with pytest.raises(RuntimeError, match=message) as error:
|
||||
reader_type(master_path)
|
||||
assert str(master_path) in str(error.value)
|
||||
|
||||
for index in range(2):
|
||||
(master_path.parent / f"run_d0_f{index}_0.raw").unlink()
|
||||
with pytest.raises(RuntimeError, match=message) as error:
|
||||
reader_type(master_path)
|
||||
assert str(master_path) in str(error.value)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("method, args, kwargs", [
|
||||
("read_frame", (), {}),
|
||||
("read_n", (1,), {}),
|
||||
("read_roi", (0,), {}),
|
||||
("read_rois", (), {}),
|
||||
("read_n_with_roi", (1,), {"roi_index": 0}),
|
||||
("frame_number", (2,), {}),
|
||||
])
|
||||
def test_raw_read_error_context(small_raw_file, method, args, kwargs):
|
||||
reader = RawFile(small_raw_file)
|
||||
reader.seek(2)
|
||||
with pytest.raises(RuntimeError) as error:
|
||||
getattr(reader, method)(*args, **kwargs)
|
||||
assert "frame index 2" in str(error.value)
|
||||
assert str(small_raw_file) in str(error.value)
|
||||
assert ".raw" not in str(error.value)
|
||||
if method == "read_frame":
|
||||
assert "file contains 2 frames (indices are zero-based)" in str(error.value)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("frame_index", [2, 17])
|
||||
def test_generic_raw_read_error_context(small_raw_file, frame_index):
|
||||
reader = File(small_raw_file)
|
||||
with pytest.raises(RuntimeError) as error:
|
||||
reader.read_frame(frame_index)
|
||||
assert f"frame index {frame_index}" in str(error.value)
|
||||
assert str(small_raw_file) in str(error.value)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("raw_subfile", [False, True])
|
||||
def test_truncated_raw_read_error_context(small_raw_file, raw_subfile):
|
||||
data_path = small_raw_file.parent / "run_d0_f1_0.raw"
|
||||
if raw_subfile:
|
||||
reader = RawSubFile(
|
||||
small_raw_file.parent / "run_d0_f0_0.raw",
|
||||
DetectorType.Jungfrau, 2, 3, 16,
|
||||
)
|
||||
else:
|
||||
reader = RawFile(small_raw_file)
|
||||
data_path.write_bytes(bytes(112))
|
||||
reader.seek(1)
|
||||
with pytest.raises(RuntimeError) as error:
|
||||
reader.read_frame()
|
||||
assert "frame index 1" in str(error.value)
|
||||
assert str(data_path) in str(error.value)
|
||||
assert "End of file" in str(error.value)
|
||||
assert "master" not in str(error.value)
|
||||
assert "RawSubFile.cpp" not in str(error.value)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("method, args", [
|
||||
("read_frame", ()),
|
||||
("read_n", (1,)),
|
||||
("read_roi", (0,)),
|
||||
("frame_number", (1,)),
|
||||
])
|
||||
def test_short_raw_subfile_sets_read_bounds(small_raw_file, method, args):
|
||||
(small_raw_file.parent / "run_d0_f1_0.raw").unlink()
|
||||
reader = RawFile(small_raw_file)
|
||||
reader.seek(1)
|
||||
with pytest.raises(RuntimeError) as error:
|
||||
getattr(reader, method)(*args)
|
||||
assert reader.total_frames == 1
|
||||
assert "frame index 1" in str(error.value)
|
||||
assert str(small_raw_file) in str(error.value)
|
||||
assert ".raw" not in str(error.value)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("counts, master_count, warn", [
|
||||
([2], 2, False),
|
||||
([2, 2], 2, False),
|
||||
([2, 2], 0, True),
|
||||
([2, 2], 1, True),
|
||||
([2, 2], 5, True),
|
||||
([1, 2], 2, True),
|
||||
([2, 1], 2, True),
|
||||
([1, 2], 1, True),
|
||||
([0, 2], 2, True),
|
||||
([0, 0], 0, False),
|
||||
])
|
||||
def test_raw_subfile_frame_counts(small_raw_file, capfd, counts, master_count, warn):
|
||||
metadata = json.loads(small_raw_file.read_text())
|
||||
metadata["Geometry"]["y"] = len(counts)
|
||||
metadata["Frames in File"] = master_count
|
||||
metadata["Total Frames"] = 1000
|
||||
small_raw_file.write_text(json.dumps(metadata))
|
||||
data = (small_raw_file.parent / "run_d0_f0_0.raw").read_bytes()
|
||||
for module, count in enumerate(counts):
|
||||
for index in range(2):
|
||||
path = small_raw_file.parent / f"run_d{module}_f{index}_0.raw"
|
||||
if index < count:
|
||||
path.write_bytes(data)
|
||||
elif index == 0:
|
||||
path.write_bytes(b"")
|
||||
else:
|
||||
path.unlink(missing_ok=True)
|
||||
|
||||
reader = RawFile(small_raw_file)
|
||||
warning = capfd.readouterr().err
|
||||
if warn:
|
||||
assert warning.count("WARNING") == 1
|
||||
assert str(small_raw_file) in warning
|
||||
else:
|
||||
assert "WARNING" not in warning
|
||||
|
||||
assert reader.total_frames == min(counts)
|
||||
assert len(reader) == min(counts)
|
||||
assert reader.master.frames_in_file == master_count
|
||||
if min(counts):
|
||||
_, frames = reader.read()
|
||||
assert frames.shape == (min(counts), 2 * len(counts), 3)
|
||||
reader.seek(0)
|
||||
_, frames = reader.read_n(10)
|
||||
assert len(frames) == min(counts)
|
||||
assert reader.frame_number(min(counts) - 1) == 0
|
||||
with pytest.raises(RuntimeError):
|
||||
reader.frame_number(min(counts))
|
||||
reader.seek(min(counts))
|
||||
with pytest.raises(RuntimeError):
|
||||
reader.read_frame()
|
||||
assert "WARNING" not in capfd.readouterr().err
|
||||
|
||||
|
||||
@pytest.mark.parametrize("short_module", [0, 1])
|
||||
def test_raw_frame_count_across_rois(small_raw_file, short_module):
|
||||
metadata = json.loads(small_raw_file.read_text())
|
||||
metadata.update({
|
||||
"Version": 8.0,
|
||||
"Geometry": {"x": 1, "y": 2},
|
||||
"Image Size": 12,
|
||||
"Receiver Rois": [
|
||||
{"xmin": 0, "xmax": 2, "ymin": 0, "ymax": 1},
|
||||
{"xmin": 0, "xmax": 2, "ymin": 2, "ymax": 3},
|
||||
],
|
||||
})
|
||||
small_raw_file.write_text(json.dumps(metadata))
|
||||
for index in range(2):
|
||||
data = (small_raw_file.parent / f"run_d0_f{index}_0.raw").read_bytes()
|
||||
(small_raw_file.parent / f"run_d1_f{index}_0.raw").write_bytes(data)
|
||||
(small_raw_file.parent / f"run_d{short_module}_f1_0.raw").unlink()
|
||||
|
||||
reader = RawFile(small_raw_file)
|
||||
assert reader.total_frames == 1
|
||||
_, rois = reader.read_rois()
|
||||
assert len(rois) == 2
|
||||
assert all(roi.shape == (2, 3) for roi in rois)
|
||||
for roi_index in range(2):
|
||||
reader.seek(0)
|
||||
_, frames = reader.read_n_with_roi(10, roi_index=roi_index)
|
||||
assert frames.shape == (1, 2, 3)
|
||||
with pytest.raises(RuntimeError):
|
||||
reader.read_roi(roi_index)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("disabled_port", [0, 1])
|
||||
def test_raw_frame_count_with_disabled_udp_port(small_raw_file, disabled_port):
|
||||
metadata = json.loads(small_raw_file.read_text())
|
||||
metadata.update({
|
||||
"Geometry": {"x": 1, "y": 2},
|
||||
"Frames in File": 0,
|
||||
"Number of UDP Interfaces": 2,
|
||||
"UDP Ports Type": ["bottom", "top"],
|
||||
"UDP Ports Disabled": [disabled_port],
|
||||
})
|
||||
small_raw_file.write_text(json.dumps(metadata))
|
||||
data = (small_raw_file.parent / "run_d0_f0_0.raw").read_bytes()
|
||||
for index in range(2):
|
||||
(small_raw_file.parent / f"run_d0_f{index}_0.raw").unlink()
|
||||
path = small_raw_file.parent / f"run_d{1 - disabled_port}_f{index}_0.raw"
|
||||
path.write_bytes(data)
|
||||
|
||||
reader = RawFile(small_raw_file)
|
||||
assert reader.total_frames == 2
|
||||
assert reader.master.frames_in_file == 0
|
||||
assert reader.frame_number(1) == 0
|
||||
_, frames = reader.read()
|
||||
assert frames.shape == (2, 2, 3)
|
||||
|
||||
|
||||
def test_raw_frame_count_without_selected_subfiles(small_raw_file):
|
||||
metadata = json.loads(small_raw_file.read_text())
|
||||
metadata.update({"Version": 8.0, "Image Size": 12, "Receiver Rois": []})
|
||||
small_raw_file.write_text(json.dumps(metadata))
|
||||
with pytest.raises(RuntimeError, match="No raw subfiles selected"):
|
||||
RawFile(small_raw_file)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("lagging_module", [0, 1])
|
||||
def test_raw_synchronization_error_identifies_subfile(
|
||||
small_raw_file, lagging_module
|
||||
):
|
||||
metadata = json.loads(small_raw_file.read_text())
|
||||
metadata["Geometry"]["y"] = 2
|
||||
small_raw_file.write_text(json.dumps(metadata))
|
||||
for index in range(2):
|
||||
data = (small_raw_file.parent / f"run_d0_f{index}_0.raw").read_bytes()
|
||||
(small_raw_file.parent / f"run_d1_f{index}_0.raw").write_bytes(data)
|
||||
|
||||
ahead_path = small_raw_file.parent / f"run_d{1 - lagging_module}_f1_0.raw"
|
||||
with ahead_path.open("r+b") as output:
|
||||
output.write(np.array([1], dtype=np.uint64).tobytes())
|
||||
|
||||
reader = RawFile(small_raw_file)
|
||||
reader.seek(1)
|
||||
with pytest.raises(RuntimeError) as error:
|
||||
reader.read_frame()
|
||||
assert "frame index 1" in str(error.value)
|
||||
assert "synchroniz" in str(error.value)
|
||||
lagging_path = small_raw_file.parent / f"run_d{lagging_module}_f1_0.raw"
|
||||
assert str(lagging_path) in str(error.value)
|
||||
|
||||
@pytest.mark.withdata
|
||||
def test_read_rawfile_with_roi_spanning_over_one_module(test_data_path):
|
||||
|
||||
@@ -110,3 +401,41 @@ def test_read_rawfile_eiger_and_compare_to_numpy(test_data_path):
|
||||
header, image1 = f.read_frame()
|
||||
|
||||
assert (image == image1).all()
|
||||
|
||||
|
||||
@pytest.mark.withdata
|
||||
def test_read_eiger_udp_port_disabled(test_data_path):
|
||||
with RawFile(test_data_path / "raw/eiger/one_udp_port_disabled_master_0.json") as f:
|
||||
_, frame = f.read_rois()
|
||||
|
||||
assert(len(frame) == 2)
|
||||
assert frame[0].shape == (256, 512)
|
||||
assert frame[1].shape == (256, 1024)
|
||||
assert f.master.udp_port_types == [UDPPortPosition.LEFT, UDPPortPosition.RIGHT]
|
||||
rois = f.master.rois
|
||||
assert len(rois) == 2
|
||||
assert rois[0] == ROI(512, 1024, 0, 256)
|
||||
assert rois[1] == ROI(0, 1024, 256, 512)
|
||||
|
||||
with RawFile(test_data_path / "raw/eiger/quad_eiger_disabled_bottom_port_master_0.json") as f:
|
||||
_, frame = f.read_frame()
|
||||
|
||||
assert frame.shape == (256, 512)
|
||||
assert(f.master.disabled_udp_ports == [1])
|
||||
assert f.master.udp_port_types == [UDPPortPosition.TOP, UDPPortPosition.BOTTOM]
|
||||
rois = f.master.rois
|
||||
assert len(rois) == 1
|
||||
assert rois[0] == ROI(0, 512, 256, 512)
|
||||
|
||||
with RawFile(test_data_path / "raw/eiger/2_modules_eiger_disabled_udp_port_master_0.json") as f:
|
||||
_, frame = f.read_rois()
|
||||
|
||||
assert(len(frame) == 2)
|
||||
assert frame[0].shape == (512, 512)
|
||||
assert frame[1].shape == (512, 512)
|
||||
assert (f.master.disabled_udp_ports == [1, 3, 5, 7])
|
||||
assert f.master.udp_port_types == [UDPPortPosition.LEFT, UDPPortPosition.RIGHT]
|
||||
rois = f.master.rois
|
||||
assert len(rois) == 2
|
||||
assert rois[0] == ROI(0, 512, 0, 512)
|
||||
assert rois[1] == ROI(1024, 1536, 0, 512)
|
||||
@@ -1,7 +1,9 @@
|
||||
|
||||
|
||||
import json
|
||||
|
||||
import pytest
|
||||
from aare import RawMasterFile, ReadoutMode, DetectorType
|
||||
from aare import RawMasterFile, ReadoutMode, DetectorType, FrameDiscardPolicy
|
||||
|
||||
|
||||
@pytest.mark.withdata
|
||||
@@ -11,4 +13,37 @@ def test_read_rawfile_quad_eiger_and_compare_to_numpy(test_data_path):
|
||||
|
||||
f = RawMasterFile(file_name)
|
||||
assert(f.reading_mode == ReadoutMode.UNKNOWN)
|
||||
assert(f.detector_type == DetectorType.Jungfrau)
|
||||
assert(f.detector_type == DetectorType.Jungfrau)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"policy, expected_policy",
|
||||
[
|
||||
("nodiscard", FrameDiscardPolicy.NoDiscard),
|
||||
("discard", FrameDiscardPolicy.Discard),
|
||||
("discardpartial", FrameDiscardPolicy.DiscardPartial),
|
||||
],
|
||||
)
|
||||
def test_raw_master_file_context_manager(tmp_path, policy, expected_policy):
|
||||
file_name = tmp_path / "run_master_0.json"
|
||||
file_name.write_text(json.dumps({
|
||||
"Version": 7.2,
|
||||
"Detector Type": "Jungfrau",
|
||||
"Timing Mode": "auto",
|
||||
"Geometry": {"x": 1, "y": 1},
|
||||
"Image Size in bytes": 12,
|
||||
"Pixels": {"x": 3, "y": 2},
|
||||
"Max Frames Per File": 1,
|
||||
"Total Frames": 2,
|
||||
"Frames in File": 2,
|
||||
"Frame Padding": 1,
|
||||
"Frame Discard Policy": policy,
|
||||
}))
|
||||
|
||||
|
||||
with RawMasterFile(file_name) as context_file:
|
||||
assert context_file.reading_mode == ReadoutMode.UNKNOWN
|
||||
assert context_file.detector_type == DetectorType.Jungfrau
|
||||
frame_discard_policy = context_file.frame_discard_policy
|
||||
assert isinstance(frame_discard_policy, FrameDiscardPolicy)
|
||||
assert frame_discard_policy == expected_policy
|
||||
@@ -0,0 +1,74 @@
|
||||
# SPDX-License-Identifier: MPL-2.0
|
||||
"""Probe make_view in np_helper.hpp through the bindings in _aare.testing"""
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from aare._aare import testing
|
||||
|
||||
|
||||
|
||||
|
||||
def test_make_view_roundtrip():
|
||||
"""Send an array to C++ and check that we get the same array back"""
|
||||
arr = np.arange(30, dtype=np.float64).reshape(5, 6)
|
||||
out = testing.view_roundtrip_2d(arr)
|
||||
np.testing.assert_array_equal(out, arr)
|
||||
|
||||
#Sanity check to make sure we got a copy an not just the same thing back
|
||||
arr[3,2] = 31655
|
||||
assert arr[3,2] != out[3,2]
|
||||
|
||||
def test_make_view_const_roundtrip():
|
||||
"""Const works if the array is not writeable"""
|
||||
arr = np.arange(30, dtype=np.float64).reshape(5, 6)
|
||||
arr.flags.writeable = False
|
||||
out = testing.const_view_roundtrip_2d(arr)
|
||||
np.testing.assert_array_equal(out, arr)
|
||||
|
||||
def test_make_view_const_roundtrip_writable():
|
||||
"""Const works also if the array is writeable"""
|
||||
arr = np.arange(30, dtype=np.float64).reshape(5, 6)
|
||||
out = testing.const_view_roundtrip_2d(arr)
|
||||
np.testing.assert_array_equal(out, arr)
|
||||
|
||||
def test_make_view_fill():
|
||||
arr = np.arange(30, dtype=np.float64).reshape(5, 6)
|
||||
testing.view_fill_2d(arr, 7)
|
||||
assert (arr == 7).all()
|
||||
|
||||
def test_make_view_fill_rejects_nonwriteable():
|
||||
arr = np.arange(30, dtype=np.float64).reshape(5, 6)
|
||||
arr.flags.writeable = False
|
||||
with pytest.raises(ValueError, match="not writeable"):
|
||||
testing.view_fill_2d(arr, 7)
|
||||
|
||||
def test_make_view_2d_rejects_1d():
|
||||
arr = np.arange(30, dtype=np.uint16)
|
||||
with pytest.raises(ValueError, match="Expected 2D array, got 1D"):
|
||||
testing.view_roundtrip_2d(arr)
|
||||
|
||||
def test_make_view_2d_rejects_3d():
|
||||
arr = np.arange(30, dtype=np.uint16).reshape(3, 5, 2)
|
||||
with pytest.raises(ValueError, match="Expected 2D array, got 3D"):
|
||||
testing.view_roundtrip_2d(arr)
|
||||
|
||||
def test_make_view_rejects_noncontiguous():
|
||||
arr = np.arange(30, dtype=np.float64).reshape(5, 6)
|
||||
arr = arr[::2, ::3] # non-contiguous
|
||||
assert not arr.flags.c_contiguous
|
||||
with pytest.raises(ValueError, match="not C-contiguous"):
|
||||
testing.view_roundtrip_2d(arr)
|
||||
|
||||
def test_make_view_rejects_noncontiguous_const():
|
||||
arr = np.arange(30, dtype=np.float64).reshape(5, 6)
|
||||
arr = arr[::2, ::3] # non-contiguous
|
||||
assert not arr.flags.c_contiguous
|
||||
with pytest.raises(ValueError, match="not C-contiguous"):
|
||||
testing.const_view_roundtrip_2d(arr)
|
||||
|
||||
def test_make_view_rejects_fortran_order():
|
||||
arr = np.asfortranarray(np.arange(30, dtype=np.float64).reshape(5, 6))
|
||||
assert not arr.flags.c_contiguous
|
||||
with pytest.raises(ValueError, match="not C-contiguous"):
|
||||
testing.view_roundtrip_2d(arr)
|
||||
|
||||
Reference in new issue
Block a user