Source code for ophyd_async.core._flyer

from abc import ABC, abstractmethod
from typing import Dict, Generic, Sequence

from bluesky.protocols import DataKey, Flyable, Preparable, Reading, Stageable

from ._device import Device
from ._signal import SignalR
from ._status import AsyncStatus
from ._utils import T, merge_gathered_dicts


[docs] class TriggerLogic(ABC, Generic[T]):
[docs] @abstractmethod async def prepare(self, value: T): """Move to the start of the flyscan"""
[docs] @abstractmethod async def kickoff(self): """Start the flyscan"""
[docs] @abstractmethod async def complete(self): """Block until the flyscan is done"""
[docs] @abstractmethod async def stop(self): """Stop flying and wait everything to be stopped"""
[docs] class StandardFlyer( Device, Stageable, Preparable, Flyable, Generic[T], ): def __init__( self, trigger_logic: TriggerLogic[T], configuration_signals: Sequence[SignalR] = (), name: str = "", ): self._trigger_logic = trigger_logic self._configuration_signals = tuple(configuration_signals) super().__init__(name=name) @property def trigger_logic(self) -> TriggerLogic[T]: return self._trigger_logic
[docs] @AsyncStatus.wrap async def stage(self) -> None: await self.unstage()
[docs] @AsyncStatus.wrap async def unstage(self) -> None: await self._trigger_logic.stop()
[docs] def prepare(self, value: T) -> AsyncStatus: """Setup trajectories""" return AsyncStatus(self._prepare(value))
async def _prepare(self, value: T) -> None: # Move to start and setup the flyscan await self._trigger_logic.prepare(value)
[docs] @AsyncStatus.wrap async def kickoff(self) -> None: await self._trigger_logic.kickoff()
[docs] @AsyncStatus.wrap async def complete(self) -> None: await self._trigger_logic.complete()
[docs] async def describe_configuration(self) -> Dict[str, DataKey]: return await merge_gathered_dicts( [sig.describe() for sig in self._configuration_signals] )
[docs] async def read_configuration(self) -> Dict[str, Reading]: return await merge_gathered_dicts( [sig.read() for sig in self._configuration_signals] )