How to write a flying Device#

A flyer is a Device that hardware-triggers a scan: bluesky prepares it, kicks it off, then waits for it to complete while the hardware runs autonomously. Rather than implement the Preparable, Flyable, and Stageable protocols by hand, compose a FlyableLogic instance into a StandardFlyable. StandardFlyable drives the three phases in order and threads a context object between them:

prepare(value) → FlyableLogic.on_prepare(value) moves to the start and arms the hardware, returning a context object.
kickoff() → FlyableLogic.on_kickoff(ctx) starts the scan, returning the (possibly updated) context.
complete() → FlyableLogic.on_complete(ctx) blocks until the scan is done.

stage()/unstage() run the logic’s on_stage/on_unstage, which default to stop.

Implementing FlyableLogic#

Subclass FlyableLogic — it is generic over two types: PrepareT, the per-scan info passed to prepare(), and CtxT, the context threaded between the phases (use None if there is no state to carry). Fill in the hooks that your hardware needs:

Method	Default behaviour	Override to…
`on_prepare(value)`	abstract	move to the start, arm the hardware, and return a context
`on_kickoff(ctx)`	abstract	start the scan and return the context for `on_complete`
`on_complete(ctx)`	abstract	block until the scan finishes
`stop()`	no-op	disarm the hardware and wait for it to stop
`on_stage()`	calls `stop()`	set the hardware up on `stage()` if it differs from stop
`on_unstage()`	calls `stop()`	clean the hardware up on `unstage()` if it differs from stop

There are two ways to get a Device out of a FlyableLogic: mix StandardFlyable into a Device class (for hardware that is always a flyer, like a motor), or call FlyableLogic.with_device to wrap a bare logic object in an ephemeral flyer inside a plan (for hardware configured per scan, like a PandA block).

A movable device that also flies#

An EPICS Motor is the first case: it is a StandardMovable and a StandardFlyable (and a StandardReadable). A single logic object, MotorFlyableMovableLogic, implements both MovableLogic and FlyableLogic, so its fly hooks can reuse the same move/check_move methods as a normal set(). The per-scan info is a FlyMotorInfo and the threaded context is a small dataclass that carries the move status from on_kickoff to on_complete:

@dataclass
class MotorFlyCtx:
    """State threaded through a motor fly scan (prepare -> kickoff -> complete)."""

    #: The info the scan was prepared with, produced by `on_prepare`.
    fly_info: FlyMotorInfo
    #: The move to the end position, produced by `on_kickoff`.
    status: AsyncStatus | None = None

on_prepare validates the requested velocity against the motor’s limits, moves to the run-up start position, and returns the context:

    async def on_prepare(self, value: FlyMotorInfo) -> MotorFlyCtx:
        """Move to the beginning of a run-up distance ready for a fly scan."""
        # Velocity at which the motor travels from start to end, in motor egu/s.
        max_speed, egu = await asyncio.gather(
            self.max_velocity.get_value(), self.motor_egu.get_value()
        )
        if value.speed > max_speed:
            raise MotorLimitsError(
                f"Speed {value.speed} {egu}/s was requested for motor "
                f"{self.readback.name} with max speed of {max_speed} {egu}/s."
            )
        # Check the run-up and run-down positions are within limits
        acceleration_time = await self.acceleration_time.get_value()
        ramp_up_start_pos = value.ramp_up_start_pos(acceleration_time)
        ramp_down_end_pos = value.ramp_down_end_pos(acceleration_time)
        await asyncio.gather(
            self.check_move(ramp_up_start_pos), self.check_move(ramp_down_end_pos)
        )
        # Move to the run-up start at maximum velocity
        await self.velocity.set(abs(max_speed))
        old_position = await self.readback.get_value()
        timeout = await self.calculate_timeout(old_position, ramp_up_start_pos)
        await self.move(ramp_up_start_pos, lambda: timeout)
        # Set the velocity we will use for the fly scan
        await self.velocity.set(value.speed)
        return MotorFlyCtx(fly_info=value)

on_kickoff starts the move to the far end of the run-down and stashes the resulting status on the context so on_complete can await it:

    async def on_kickoff(self, ctx: MotorFlyCtx) -> MotorFlyCtx:
        """Begin moving the motor from the prepared position to the final position."""
        acceleration_time = await self.acceleration_time.get_value()
        target = ctx.fly_info.ramp_down_end_pos(acceleration_time)
        if ctx.fly_info.timeout == CALCULATE_TIMEOUT:
            initial = await self.readback.get_value()
            timeout = await self.calculate_timeout(initial, target)
        else:
            timeout = ctx.fly_info.timeout
        ctx.status = AsyncStatus(self.move(target, lambda: timeout))
        # Wait out the run-up so the motor is at constant velocity before
        # kickoff() returns. A plan can then kickoff the motor and afterwards
        # kickoff internally-triggered detectors, which is more accurate for
        # long acceleration times.
        await asyncio.sleep(acceleration_time)
        return ctx

on_complete simply blocks on that status:

    async def on_complete(self, ctx: MotorFlyCtx) -> None:
        """Block until the motor reaches the fly-scan end position."""
        status = error_if_none(
            ctx.status, f"kickoff for motor {self.readback.name} not called."
        )
        await status

Because the logic is also a MovableLogic, complete() reports progress to watchers (e.g. progress bars) automatically by observing the readback — reusing the same WatcherUpdate machinery as StandardMovable.set. A flyer whose logic is not movable (see the next section) just blocks with no progress updates. See When should a device extend movable for more on MovableLogic.

Wire the logic into the Device with a @cached_property. Motor builds one shared instance and returns it from both movable_logic and flyable_logic so the move and fly paths act on the same signals:

    @cached_property
    def _logic(self) -> MotorFlyableMovableLogic:
        """The combined move + fly logic, shared by movable_logic and flyable_logic."""
        return MotorFlyableMovableLogic(
            readback=self.user_readback,
            setpoint=self.user_setpoint,
            motor_stop=self.motor_stop,
            low_limit_travel=self.low_limit_travel,
            high_limit_travel=self.high_limit_travel,
            dial_low_limit_travel=self.dial_low_limit_travel,
            dial_high_limit_travel=self.dial_high_limit_travel,
            velocity=self.velocity,
            acceleration_time=self.acceleration_time,
            motor_done_move=self.motor_done_move,
            max_velocity=self.max_velocity,
            motor_egu=self.motor_egu,
        )

@cached_property
def movable_logic(self) -> MotorFlyableMovableLogic:
    return self._logic

@cached_property
def flyable_logic(self) -> MotorFlyableMovableLogic:
    return self._logic

An ephemeral flyer created in a plan#

Some hardware is only a flyer for the duration of a scan and is configured differently each time. The PandA position-compare block is an example: StaticPcompFlyableLogic wraps a single PcompBlock and carries no state between phases, so it is a FlyableLogic[PcompInfo, None]. Its on_prepare disarms then programs the block from the per-scan PcompInfo:

    async def on_prepare(self, value: PcompInfo):
        await self.pcomp.enable.set(PandaBitMux.ZERO)
        await asyncio.gather(
            self.pcomp.start.set(value.start_position),
            self.pcomp.width.set(value.pulse_width),
            self.pcomp.step.set(value.rising_edge_step),
            self.pcomp.pulses.set(value.number_of_pulses),
            self.pcomp.dir.set(value.direction),
        )

on_kickoff enables the block and waits for it to go active, while on_complete waits for it to go inactive again (the context is None, so it is ignored):

    async def on_kickoff(self, ctx: None) -> None:
        await self.pcomp.enable.set(PandaBitMux.ONE)
        await wait_for_value(self.pcomp.active, True, timeout=1)

    async def on_complete(self, ctx: None) -> None:
        await wait_for_value(self.pcomp.active, False, timeout=None)

stop() disarms the block; it is called by the default on_stage/on_unstage:

    async def stop(self):
        await self.pcomp.enable.set(PandaBitMux.ZERO)
        await wait_for_value(self.pcomp.active, False, timeout=1)

Rather than declare this logic on a Device class, construct it in the plan from the PandA block you want to drive and call FlyableLogic.with_device to get a ready flyer to prepare, kickoff, and complete:

flyer = StaticPcompFlyableLogic(panda.pcomp[1]).with_device(name="pcomp")