Scan plans and execution

ScanPlan and ScanExecutor drive DataFrame-based scans: plan from a DataFrame, execute with optional detector and RunWriter.

class resonance.api.core.scan.ScanPlan(points: list[ScanPoint], motor_names: list[str], ai_channels: list[str], shutter: str = 'Light Output', actuate_every: bool = False)

Bases: object

Validated scan plan constructed from a list of points or a DataFrame.

Parameters:

• points (list[ScanPoint]) – Ordered sequence of scan points.

• motor_names (list[str]) – Names of all motors referenced in the scan.

• ai_channels (list[str]) – Analog input channel names to acquire at each point.

• shutter (str) – DIO channel name of the light shutter.

• actuate_every (bool) – If True, the shutter is opened and closed per point. If False, the shutter is opened once for the entire scan.

Examples

>>> df = pd.DataFrame({
...     "Sample X": [0, 10, 20],
...     "Sample Y": [0, 0, 0],
...     "exposure": [0.1, 0.1, 0.1],
... })
>>> scan_plan = ScanPlan.from_dataframe(df, ai_channels=["Photodiode"])

describe(motor_time: float = 0.1, api_time: float = 0.5) → None

Print a human-readable summary of the scan plan.

Displays point count, unique motor values, and estimated duration.

Parameters:

• motor_time (float) – Motor-move overhead per point in seconds used for the estimate.

• api_time (float) – API overhead per point in seconds used for the estimate.

estimated_duration_seconds(motor_time: float = 0.1, api_time: float = 0.5) → float

Compute estimated total scan duration.

Per-point cost: motor_time + api_time + exposure_time + delay_after_move.

Parameters:

• motor_time (float) – Expected motor-move overhead per point in seconds.

• api_time (float) – Expected API round-trip overhead per point in seconds.

Returns:

Total estimated duration in seconds.

Return type:

float

Notes

Does not account for shutter actuation overhead when actuate_every=True.

classmethod from_dataframe(df: DataFrame, ai_channels: list[str] | None = None, default_exposure: float = 1.0, default_delay: float = 0.2, shutter: str = 'Light Output', actuate_every: bool = False) → ScanPlan

Build a validated scan plan from a DataFrame.

Each row becomes one ScanPoint. Motor columns are detected via validate_scan_dataframe; an optional exposure column is also detected.

Parameters:

• df (pd.DataFrame) – DataFrame whose columns are motor names and optionally an exposure time column.

• ai_channels (list[str] or None) – Channels to acquire at each point. If None, a beamline default set is used.

• default_exposure (float) – Exposure time in seconds applied when the DataFrame has no exposure column.

• default_delay (float) – Settle delay in seconds after each motor move.

• shutter (str) – DIO channel name of the light shutter.

• actuate_every (bool) – Per-point shutter mode.

Returns:

Fully validated scan plan.

Return type:

ScanPlan

Raises:

ValidationError – If the DataFrame fails structural or value validation.

class resonance.api.core.scan.ScanExecutor(conn: BCSz.BCSServer)

Bases: object

Executes ScanPlan instances against a live BCS server.

Supports two interrupt modes:

1. Programmatic abort: call await executor.abort() from any async context to set the abort flag. The next check inside execute_point or wait_for_settle raises ScanAbortedError.

2. Jupyter / IPython interrupt: create the scan as an asyncio.Task and call await bl.abort_scan() from another cell to set the abort flag and stop after the current point.

asyncio.CancelledError is raised when the Task is cancelled via task.cancel(). execute_scan catches this, sets the abort flag, and returns any partial results already collected. If no results have been collected the error is re-raised.

Parameters:

conn (BCSz.BCSServer) – Active BCS server connection.

async abort() → None

Request abort of the running scan.

Sets the internal AbortFlag. The scan loop will raise ScanAbortedError at the next abort-check site (start of execute_point, inside wait_for_settle, or inside wait_for_motors).

Notes

Safe to call from a separate task or thread-pool executor while the scan is running.

property current_scan: ScanPlan | None

Currently running scan plan.

Returns:

The active ScanPlan during execution, or None when idle.

Return type:

ScanPlan or None

async execute_point(point: ScanPoint, motor_timeout: float = 30.0, restore_motors: bool = False, use_shutter: bool = True, detector: AreaDetector | None = None) → ScanResult

Execute a single scan point.

Parameters:

• point (ScanPoint) – The point to execute.

• motor_timeout (float) – Maximum time in seconds to wait for motors to reach position.

• restore_motors (bool) – If True, motor positions are restored to their pre-move values after the point completes.

• use_shutter (bool) – If True, the shutter is opened and closed around acquisition. Pass False when the caller already holds the shutter open.

• detector (AreaDetector or None, optional) – If provided, a 2D detector image is acquired after AI acquisition using the point’s exposure_time. Shutter actuation is hardware-driven; no plan-level shutter wraps this call.

Returns:

Measured values with per-channel ufloat statistics.

Return type:

ScanResult

Raises:

• ScanAbortedError – If the abort flag is set before or during execution.

• MotorError – If a motor move or restore fails.

async execute_scan(scan_plan: ScanPlan, progress: bool = True, writer: RunWriter | None = None, detector: AreaDetector | None = None) → pd.DataFrame

Execute a complete scan plan and return results as a DataFrame.

The shutter behaviour depends on scan_plan.actuate_every:

• False (default): the shutter is opened once before the first point and closed after the last point (or on abort).

• True: the shutter is actuated individually for every point via execute_point.

Interrupt modes

Programmatic: call await executor.abort() from any async context. The AbortFlag is checked at the start of each point; ScanAbortedError is raised and partial results are returned.

Jupyter / IPython: create this coroutine as an asyncio.Task and call await bl.abort_scan() from another cell to set the abort flag and stop after the current point.

asyncio.CancelledError: raised when the Task is cancelled via task.cancel(). Partial results are returned if any points completed; otherwise the error is re-raised.

param scan_plan:

Validated scan plan to execute.

type scan_plan:

ScanPlan

param progress:

If True and tqdm is installed, show an async progress bar. Falls back to simple per-point print statements.

type progress:

bool

param writer:

If provided, scalar scan data (motor positions, AI means, exposure, timestamps) are written to the open beamtime database. The writer must not have an open run before this method is called; it will call open_run, open_stream, write_event, and close_run internally.

type writer:

RunWriter or None, optional

param detector:

If provided, a 2D image is acquired at each scan point and written to the “detector_image” field in the primary stream. Requires writer to be set for persistence. Shutter is hardware-driven.

type detector:

AreaDetector or None, optional

returns:

One row per completed point with motor positions, per-channel mean/std columns, exposure time, and timestamp.

rtype:

pd.DataFrame

Notes

# TODO: add Bluesky-compatible start/stop document emission once RunWriter is stable