Source code for blissoda.demo.tests.itest_stop_scan

import numpy

from ...bliss_globals import setup_globals
from ...version_utils import has_minimal_version
from .. import testing
from ..processors.stop_scan import DemoPreset




@testing.integration_test
def test_stop_scan():
    if not has_minimal_version("bliss", "2.2"):
        testing.skip_integration_test("Requires bliss>=2.2")

    expo = 0.2
    detectors = [setup_globals.diode1, setup_globals.difflab6]

    # Stop criterium is based the total sum of all data of several detectors.
    # Get a reasonable limit for the total.
    scan = setup_globals.ct(expo, *detectors)
    max_difflab6 = numpy.nanmax(scan.streams["difflab6:image"][:])
    max_diode1 = numpy.nanmax(
        scan.streams["simulation_diode_sampling_controller:diode1"][:]
    )
    lima_limits = {"difflab6": float(max_difflab6 * 5)}
    scalar_counter_limits = {"diode1": float(max_diode1 * 5)}

    # Scan with preset to start/stop data processing and handle feedback streamed through Redis.
    p = DemoPreset(lima_limits, scalar_counter_limits)
    scan = setup_globals.loopscan(100, expo, *detectors, run=False)
    scan.acq_chain.add_preset(p)
    scan.run()

    # Check that the scan stopped early
    for name in ("difflab6:image", "simulation_diode_sampling_controller:diode1"):
        npoints = len(scan.streams[name])
        assert (
            npoints < 20
        ), f"Stream {name:!r} has {npoints} which is more than expected"