import numpy
from ...bliss_globals import (
setup_globals, # pyright: ignore[reportAttributeAccessIssue]
)
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_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.
"""
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}"
)