import numpy
from ...bliss_globals import (
setup_globals, # pyright: ignore[reportAttributeAccessIssue]
)
from ...utils.directories import get_dataset_processed_dir
from ...version_utils import has_minimal_version
from .. import testing
from ..processors.stop_scan_xrpd_id31 import DemoStopIntegrateSum
from ..processors.stop_scan_xrpd_id31 import id31_patching
def _measure_baseline_imax(expo: float = 0.2) -> float:
"""Return the max pixel value from a single ct with no attenuation."""
# id31_patching:
# - ensures setup_globals.atten is available;
# - ensures high energy so that attenuation with SiO2 performs as expected
# this is the id31.attenuator.Attenuator (`setup_globals.atten` on the beamline)
# *note* it is called again in DemoStopIntegrateSum
id31_patching(energy=75.05)
setup_globals.atten.bits = 0
scan = setup_globals.ct(expo, setup_globals.difflab6)
return float(numpy.nanmax(scan.streams["difflab6:image"][:]))
[docs]
@testing.integration_test
def test_attenuation_increase():
"""Verify that the preset increases attenuation when frames are saturating.
Strategy: set ``detector_saturation`` below the measured baseline intensity
so every frame appears saturating to the preset. The workflow threshold is
set astronomically high so the Ewoks workflow never fires a stop event and
the scan runs to normal completion. After ``npoints`` frames the preset
must have moved ``atten.bits`` above 0.
"""
if not has_minimal_version("bliss", "2.2"):
testing.skip_integration_test("Requires bliss>=2.2")
detector = setup_globals.difflab6
expo = 0.2
npoints = 5
baseline_imax = _measure_baseline_imax(expo)
# saturating threshold: set saturation well below the real detector output
saturation = baseline_imax * 0.5
p = DemoStopIntegrateSum(
workflow_threshold=1e-15, # never trigger stop
detector_name=detector.name,
detector_saturation=saturation,
frame_target_max=saturation * 0.9,
frame_target_min=saturation * 0.1,
attenuation_mode="reactive",
)
setup_globals.atten.bits = 0 # start with no attenuation
scan = setup_globals.loopscan(npoints, expo, detector, run=False, save=False)
scan.acq_chain.add_preset(p)
scan.run()
assert (
setup_globals.atten.bits > 0
), f"Expected atten.bits > 0 after saturating frames, got {setup_globals.atten.bits}"
[docs]
@testing.integration_test
def test_attenuation_decrease():
"""Verify that the preset decreases attenuation when frames are too dim.
Strategy: start with high attenuation (``bits = 10``) and set
``frame_target_min`` well above the measured baseline intensity so every
frame appears too dim. ``detector_saturation`` is set high enough that
removing attenuation is predicted to be safe. The preset must lower
``atten.bits`` below the initial value.
"""
if not has_minimal_version("bliss", "2.2"):
testing.skip_integration_test("Requires bliss>=2.2")
detector = setup_globals.difflab6
expo = 0.2
npoints = 5
initial_bits = 10
baseline_imax = _measure_baseline_imax(expo)
# dim threshold: target_min above what the detector actually returns
saturation = baseline_imax * 20
p = DemoStopIntegrateSum(
workflow_threshold=1e-15, # never trigger stop
detector_name=detector.name,
detector_saturation=saturation,
frame_target_max=saturation * 0.9,
frame_target_min=baseline_imax * 2, # above measured baseline: always too dim
attenuation_mode="reactive",
)
setup_globals.atten.bits = initial_bits
scan = setup_globals.loopscan(npoints, expo, detector, run=False, save=False)
scan.acq_chain.add_preset(p)
scan.run()
assert setup_globals.atten.bits < initial_bits, (
f"Expected atten.bits < {initial_bits} after dim frames, "
f"got {setup_globals.atten.bits}"
)
[docs]
@testing.integration_test
def test_scan_stop_pyfai():
if not has_minimal_version("bliss", "2.2"):
testing.skip_integration_test("Requires bliss>=2.2")
# set detector object and exposure time
detector = setup_globals.difflab6
expo = 0.2
# create the preset; the Demo version sets up:
# - the ID31 mock attenuator,
# - demo PyFAI config,
# - demo Ewoks workflow
p = DemoStopIntegrateSum(workflow_threshold=0.005, detector_name=detector.name)
# run a long loopscan with the preset - should stop at 10-15 frames
scan = setup_globals.loopscan(100, expo, detector, run=False)
scan.acq_chain.add_preset(p)
scan.run()
# check that the scan stopped early as expected
npoints = len(scan.streams["difflab6:image"])
assert npoints < 25, f"Stream 'difflab6:image' has {npoints} points, expected < 25"
[docs]
@testing.integration_test
def test_sum_frame_max_threshold_stop():
"""The accumulated-frame max-counts criterion stops the scan early and
the produced ``_sum.h5`` records which threshold fired and at what value.
Strategy: set ``sum_frame_max_threshold`` slightly above the per-frame
baseline so the accumulated sum trips after a handful of frames, and
pin ``workflow_threshold`` (rel_err) to its sentinel so the acc_profile
criterion cannot fire first. Then inspect the ``_sum.h5`` metadata.
"""
if not has_minimal_version("bliss", "2.2"):
testing.skip_integration_test("Requires bliss>=2.2")
from pathlib import Path
import h5py
detector = setup_globals.difflab6
expo = 0.2
baseline_imax = _measure_baseline_imax(expo)
sum_threshold = baseline_imax * 1.5
p = DemoStopIntegrateSum(
workflow_threshold=1e-15, # sentinel: acc_profile criterion never fires
detector_name=detector.name,
sum_frame_max_threshold=sum_threshold,
)
setup_globals.atten.bits = 0
scan = setup_globals.loopscan(100, expo, detector, run=False)
scan.acq_chain.add_preset(p)
scan.run()
npoints = len(scan.streams["difflab6:image"])
assert npoints < 100, f"Expected early stop, got full {npoints} frames"
# wait for the WriteSumFrame ewoks task to finish writing _sum.h5
if p._future is not None:
p._future.result(timeout=60)
raw_filename = scan.scan_info["filename"]
scan_nb = scan.scan_info["scan_nb"]
stem = Path(raw_filename).stem
sum_filename = Path(get_dataset_processed_dir(raw_filename)) / f"{stem}_sum.h5"
assert sum_filename.exists(), f"_sum.h5 not produced: {sum_filename}"
with h5py.File(sum_filename, "r") as f:
proc = f[f"/{scan_nb}.1/auto_stop_sum"]
assert proc["results/sum_frame_max_threshold_reached"][()]
assert not proc["results/acc_profile_threshold_reached"][()]
assert (
proc["results/sum_frame_max_value_at_stop"][()]
> proc["parameters/sum_frame_max_threshold"][()]
)
assert numpy.isnan(proc["results/acc_profile_value_at_stop"][()])
[docs]
@testing.integration_test
def test_freeze_mode_smoke():
"""Smoke-test ``attenuation_mode='freeze'``: the preset classifies using the
canned frame-0 metric, runs its setup phase, and leaves the scan in a
consistent state.
NOTE: the demo detector ``difflab6`` does *not* respond to ``atten.bits``, so the
safe-frame loop will not **converge** on this detector.
This test therefore only asserts:
- the event dispatch wires the frame-0 metric through to classification;
- ``_apply_filter`` is called at least once during setup (unless the
initial frame happens to satisfy the boundary condition);
- the scan runs to completion without error.
Reading the live log of this test you will see the predicted-optimal jump
step the attenuator by exactly +2 bits every iteration (0 → 2 → 4 → …).
That stride is a demo artefact, not a behaviour signature: this test pins
``ghost_threshold = baseline_imax × 0.5`` so a move is guaranteed, the
SiO2 transmission table drops by ≈ factor 2 per bit, and difflab6 keeps
returning the same ``imax`` regardless of atten — so the helper applies
its "strictly less than threshold" rule against a perceived count rate
that quadruples each iteration. On a real detector the actual ``imax``
would drop with attenuation and the loop would converge after the first
jump+verify.
"""
if not has_minimal_version("bliss", "2.2"):
testing.skip_integration_test("Requires bliss>=2.2")
detector = setup_globals.difflab6
expo = 0.2
npoints = 5
baseline_imax = _measure_baseline_imax(expo)
# threshold below baseline so frame 0 triggers setup moves
ghost_threshold = baseline_imax * 0.5
p = DemoStopIntegrateSum(
workflow_threshold=1e-15, # real workflow skipped via canned_metrics
detector_name=detector.name,
attenuation_mode="freeze",
ghost_threshold_per_frame=ghost_threshold,
spottiness_threshold=0.1,
metric_timeout=0.2,
canned_metrics=[{"frame": 0, "max_pixel": baseline_imax, "spottiness": 0.05}],
)
setup_globals.atten.bits = 0
scan = setup_globals.loopscan(npoints, expo, detector, run=False, save=False)
scan.acq_chain.add_preset(p)
scan.run()
assert p.classification == "standard", (
f"Expected 'standard' classification from canned spottiness=0.05, "
f"got {p.classification!r}"
)