User-specified logic callback

[Tom-Willemsen]

As a reflectometry scientist, I would like a way to perform a scan and react to it's data in some user-specified way.

Discussed 10/07/2025 1:24:48

An example use case is:

• Do a coarse scan, which gives data that does not have a 'nice' structure and will not fit using our normal fitting models
• Run user-specified logic to select an interesting region
• Re-run a fine scan over that interesting region, now using standard ibex_bluesky_core fitting models on that second scan.

It is already possible for scientists to define their own run-engine callbacks to do this. However, there is a certain amount of book-keeping logic required.

This ticket is to implement a wrapper with some simplifying assumptions to allow scientists to just write a function which operates on numpy arrays, in a simpler way than writing a full callback. We will necessarily make some simplifying assumptions (e.g. all data are scalars, and callback is only executed once at the end of a scan) - for full control, scientists can still write a fully custom callback.

The kind of thing we want to be able to write in a plan wrapper is:

from ibex_bluesky_core.callbacks import custom_callback  # to be implemented in this ticket


def some_user_specified_logic(x: npt.NDArray, y: npt.NDArray, y_err: npt.NDArray | None = None) -> Any:
    # Fiddle with the arrays in some user-specified way
    return (42, 1729)


def plan():
    motor = ...
    detector = ...

    my_callback = custom_callback(
        some_user_specified_logic, 
        x=motor.name, 
        y=detector.intensity.name, 
        y_err=detector.intensity_err.name
    )

    @bpp.subs_decorator([my_callback])
    def _inner():
        yield from scan([detector], motor, ...)

    yield from _inner()
    a, b = my_callback.result  # Whatever type was returned from some_user_specified_logic
    yield from bps.mv(motor, (a+b)/2)