### EPICS Device Setup with Ophyd-Async

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/tutorials/implementing-devices.md

Sets up a RunEngine, subscribes a callback, and initializes EPICS demo devices including a stage and a point detector. Uses a random PV prefix to avoid name clashes and starts a demo IOC in a subprocess.

```python
"""Used for tutorial `Implementing Devices`."""

# Import bluesky and ophyd
import bluesky.plan_stubs as bps  # noqa: F401
import bluesky.plans as bp  # noqa: F401
from bluesky.callbacks.best_effort import BestEffortCallback
from bluesky.run_engine import RunEngine, autoawait_in_bluesky_event_loop

from ophyd_async.core import init_devices
from ophyd_async.epics import demo, testing

# Create a run engine and make ipython use it for `await` commands
RE = RunEngine(call_returns_result=True)
autoawait_in_bluesky_event_loop()

# Add a callback for plotting
bec = BestEffortCallback()
RE.subscribe(bec)

# Start IOC with demo pvs in subprocess
prefix = testing.generate_random_pv_prefix()
ioc = demo.start_ioc_subprocess(prefix, num_channels=3)

# All Devices created within this block will be
# connected and named at the end of the with block
with init_devices():
    # Create a sample stage with X and Y motors
    stage = demo.DemoStage(f"{prefix}STAGE:")
    # Create a multi channel counter with the same number
    # of counters as the IOC
    pdet = demo.DemoPointDetector(f"{prefix}DET:", num_channels=3)
```

--------------------------------

### Install ophyd-async

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/tutorials/installation.md

Install the core ophyd-async library and its default dependencies using pip.

```bash
$ python3 -m pip install ophyd-async
```

--------------------------------

### Install ophyd-async with Extras

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/tutorials/installation.md

Install ophyd-async with specific optional dependencies for control systems or features like Channel Access, PVAccess, Tango, simulation, or demos.

```bash
$ python3 -m pip install ophyd-async[ca,sim,demo]
```

--------------------------------

### Install ophyd-async from GitHub

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/tutorials/installation.md

Install the latest development version of ophyd-async directly from its GitHub repository.

```bash
$ python3 -m pip install git+https://github.com/bluesky/ophyd-async.git
```

--------------------------------

### Check Installed Version

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/tutorials/installation.md

Verify the installed version of ophyd-async from the command line.

```bash
$ python -m ophyd_async --version
```

--------------------------------

### Python Plan to Manage Device Settings

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/how-to/put-device-back.md

This plan demonstrates how to get initial device settings, retrieve saved settings, apply them if different, perform scans, and finally restore the initial settings.

```python
provider = YamlSettingsProvider("directory_to_save_yaml_to")

def my_plan():
    # Get the current settings from the device
    initial_settings = yield from get_current_settings(device)
    # Retrieve a previously saved settings from the provider
    known_good_settings = yield from retrieve_settings(
        provider, "yaml_file_name", device
    )
    # Apply the settings that aren't at the right value to the device
    # using the stored initial_settings from above rather than querying
    # the device again
    yield from apply_settings_if_different(
        known_good_settings,
        apply_plan=apply_settings,
        current_settings=initial_settings,
    )
    # Do what we came here to do...
    yield from do_a_scan(device)
    yield from do_another_scan(device)
    # Put it back how we found it
    yield from apply_settings_if_different(
        initial_settings,
        apply_plan=apply_settings,
    )
```

--------------------------------

### ReferenceDevice Example

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/decisions/0008-signal-types.md

Demonstrates how to use a Reference to wrap a SignalRW and modify its behavior in a Device. The set method increments the value before setting it.

```python
from ophyd_async.core import Reference

class ReferenceDevice(Device):
    def __init__(self, signal: SignalRW[int], name: ""):
        self._signal_ref = Reference(signal)
        super().__init__(name=name)

    def set(self, value) -> AsyncStatus:
        return self._signal_ref().set(value + 1)
```

--------------------------------

### Instantiate ADARAVIS Detector

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/how-to/implement-ad-detector.md

Demonstrates how to import and create an ADARAVIS Detector instance. This is the basic setup for using the detector.

```python
from ophyd_async.epics import adaravis

det = adaravis.AravisDetector("PREFIX:", path_provider)
```

--------------------------------

### Constructing a SignalR with a Custom Backend

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/devices-signals-backends.md

Example of how to instantiate a SignalR with a custom control system backend. This involves passing the backend class and its parameters to the Signal constructor.

```python
my_signal = SignalR(MyControlSystemBackend(int, cs_param="something"))
```

--------------------------------

### Start IPython with ophyd-async simulation

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/tutorials/implementing-detectors.md

Launches an IPython shell with the ophyd-async simulation environment enabled, suitable for interactive testing and development.

```bash
$ ipython --matplotlib=qt6 -i -m ophyd_async.sim
Python 3.11.11 (main, Dec  4 2024, 20:38:25) [GCC 12.2.0] 
Type 'copyright', 'credits' or 'license' for more information
IPython 8.30.0 -- An enhanced Interactive Python. Type '?' for help.

In [1]: 
```

--------------------------------

### Initialize Devices with Automatic Mock Behavior

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/tutorials/writing-tests-for-devices.md

Initialize devices with mock=True to automatically use the defined DeviceMock subclasses. This avoids manual callback setup for standard mock behavior.

```python
async with init_devices(mock=True):
    motor = Motor("BLxxI-MO-TABLE-01:X")

# No manual callback setup needed - the mock behavior is already active
await motor.user_setpoint.set(50.0)
assert await motor.user_readback.get_value() == 50.0
```

--------------------------------

### Check Python Version

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/tutorials/installation.md

Verify your Python installation meets the minimum requirement of 3.11 or later.

```bash
$ python3 --version
```

--------------------------------

### Run EPICS Demo in IPython

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/tutorials/implementing-devices.md

Starts an interactive IPython session with the EPICS demo environment loaded. This is used to test and interact with EPICS-based devices.

```bash
$ ipython --matplotlib=qt6 -i -m ophyd_async.epics.demo
Python 3.11.11 (main, Dec  4 2024, 20:38:25) [GCC 12.2.0] 
Type 'copyright', 'credits' or 'license' for more information
IPython 8.30.0 -- An enhanced Interactive Python. Type '?' for help.

In [1]: 

```

--------------------------------

### Procedural Device Definition (Ophyd-Async)

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/decisions/0006-procedural-device-definitions.md

Example of a procedural device definition in ophyd-async, explicitly defining signals.

```python
class Sensor(StandardReadable):
    def __init__(self, prefix: str, name="") -> None:
        self.value = epics_signal_r(float, prefix + "Value")
        self.mode = epics_signal_rw(EnergyMode, prefix + "Mode")
        # Set name and signals for read() and read_configuration()
        self.set_readable_signals(read=[self.value], config=[self.mode])
        super().__init__(name=name)
```

--------------------------------

### Markdown Example

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/decisions/0010-docstring-format.md

Example of prose documentation written in markdown.

```markdown
# Connecting the Device
Rather than calling [](#Device.connect) yourself which would use the wrong event loop you can 
use [](#init_devices) at startup or the equivalent [plan stub](#ensure_connected). Remember to 
pass `mock=True` during testing.

```

--------------------------------

### Navigate and Run Eiger Tests

Source: https://github.com/bluesky/ophyd-async/blob/main/tests/system_tests/epics/eiger/README.md

Changes directory to the Eiger system tests and executes the start script for IOCs and tests. This assumes a Python environment with 'ophyd-async' is already loaded.

```bash
cd system_tests/epics/eiger && ./start_iocs_and_run_tests.sh
```

--------------------------------

### Export Detector Components

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/how-to/implement-ad-detector.md

Example of defining the __all__ list to export the detector class, driver IO, trigger logic, and any custom Enum types.

```python
__all__ = [
    "AravisDetector",
    "AravisDriverIO",
    "AravisTriggerLogic",
    "AravisTriggerSource",
]
```

--------------------------------

### Read-Write Derived Signal Example

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/how-to/derive-one-signal-from-others.md

Shows a read-write derived signal that can be read and set. Use this when you need to derive a value and also be able to set it, which in turn controls underlying signals.

```python
class MovableBeamstop(Device):
    """As well as reads, this one allows you to move it.

    E.g. bps.mv(beamstop.position, BeamstopPosition.IN_POSITION)
    """

    def __init__(self, name=""):
        # Raw signals
        self.x = soft_signal_rw(float)
        self.y = soft_signal_rw(float)
        # Derived signals
        self.position = derived_signal_rw(
            self._get_position, self._set_from_position, x=self.x, y=self.y
        )
        super().__init__(name=name)

    def _get_position(self, x: float, y: float) -> BeamstopPosition:
        if abs(x) < 1 and abs(y) < 2:
            return BeamstopPosition.IN_POSITION
        else:
            return BeamstopPosition.OUT_OF_POSITION

    async def _set_from_position(self, position: BeamstopPosition) -> None:
        if position == BeamstopPosition.IN_POSITION:
            await asyncio.gather(self.x.set(0), self.y.set(0))
        else:
            await asyncio.gather(self.x.set(3), self.y.set(5))

```

--------------------------------

### Aravis Trigger Logic Implementation

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/how-to/implement-ad-detector.md

Example implementation of trigger logic for Aravis GigE and USB3 cameras. Includes configuration signals, deadtime calculation, and preparation for internal, edge, and default trigger modes.

```python
from dataclasses import dataclass
from typing import Set, Dict

from ophyd_async.core import SignalR, SignalDict
from ophyd_async.epics.adcore import DetectorTriggerLogic, trigger_info_from_num_images
from ophyd_async.epics.adcore.trigger import prepare_exposures
from ophyd_async.epics.areadetector import ADBaseIO
from ophyd_async.epics.areadetector.drivers.aravis import (  # noqa
    AravisDriverIO,
    AravisTriggerSource,
    OnOff,
)

def get_camera_deadtime(model: str, override_deadtime: float | None) -> float:
    # Placeholder for actual deadtime calculation logic
    if override_deadtime is not None:
        return override_deadtime
    if "aravis" in model.lower():
        return 0.01  # Example deadtime for Aravis
    return 0.0


@dataclass
class AravisTriggerLogic(DetectorTriggerLogic):
    """Trigger logic for Aravis GigE and USB3 cameras."""

    driver: AravisDriverIO
    override_deadtime: float | None = None

    def config_sigs(self) -> Set[SignalR]:
        return {self.driver.model}

    def get_deadtime(self, config_values: Dict[SignalR, object]) -> float:
        return get_camera_deadtime(
            model=config_values[self.driver.model],
            override_deadtime=self.override_deadtime,
        )

    async def prepare_internal(self, num: int, livetime: float, deadtime: float):
        await self.driver.trigger_mode.set(OnOff.OFF)
        await prepare_exposures(self.driver, num, livetime, deadtime)

    async def prepare_edge(self, num: int, livetime: float):
        # Trigger on the rising edge of Line1
        # trigger mode must be set first and on its own!
        # Hardware race condition in Aravis firmware requires setting trigger mode
        # separately before trigger source to avoid undefined behavior.
        # https://github.com/AravisProject/aravis/issues/1045
        await self.driver.trigger_mode.set(OnOff.ON)
        await self.driver.trigger_source.set(AravisTriggerSource.LINE1)
        await prepare_exposures(self.driver, num, livetime)

    async def default_trigger_info(self):
        return await trigger_info_from_num_images(self.driver)

```

--------------------------------

### Pytest Fixture for Detector Test Setup

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/how-to/implement-ad-detector.md

Initialize the detector in mock mode for unit testing using a pytest fixture. This includes setting up mock values for necessary PVs like file_path_exists.

```python
@pytest.fixture
async def test_adaravis(
    static_path_provider: StaticPathProvider,
) -> adaravis.AravisDetector:
    async with init_devices(mock=True):
        detector = adaravis.AravisDetector(
            "PREFIX:", adcore.ADWriterFactory.hdf(static_path_provider)
        )
    writer = detector.get_plugin("hdf", adcore.NDPluginFileIO)
    set_mock_value(writer.file_path_exists, True)
    return detector
```

--------------------------------

### Run Eiger Detector Simulator

Source: https://github.com/bluesky/ophyd-async/blob/main/tests/system_tests/epics/eiger/README.md

Starts the Eiger detector simulator in a container. Ensure SELinux is disabled before running. This command mounts shared memory and temporary directories and uses the host network.

```bash
podman run --rm -it -v /dev/shm:/dev/shm -v /tmp:/tmp --net=host ghcr.io/diamondlightsource/eiger-detector-runtime:1.16.0beta5
```

--------------------------------

### Initialize Ophyd-Async Demo Environment

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/tutorials/using-devices.md

Sets up the RunEngine, callbacks, and simulated devices for the tutorial. This code block should be run in an interactive IPython session.

```python
"""Used for tutorial `Using Devices`."""

# Import bluesky and ophyd
from pathlib import PurePath
from tempfile import mkdtemp

import bluesky.plan_stubs as bps  # noqa: F401
import bluesky.plans as bp  # noqa: F401
import bluesky.preprocessors as bpp  # noqa: F401
from bluesky.callbacks.best_effort import BestEffortCallback
from bluesky.run_engine import RunEngine, autoawait_in_bluesky_event_loop

from ophyd_async import sim
from ophyd_async.core import StaticPathProvider, UUIDFilenameProvider, init_devices

# Create a run engine and make ipython use it for `await` commands
RE = RunEngine(call_returns_result=True)
autoawait_in_bluesky_event_loop()

# Add a callback for plotting
bec = BestEffortCallback()
RE.subscribe(bec)

# Make a pattern generator that uses the motor positions
# to make a test pattern. This simulates the real life process
# of X-ray scattering off a sample
pattern_generator = sim.PatternGenerator()

# Make a path provider that makes UUID filenames within a static
# temporary directory
path_provider = StaticPathProvider(UUIDFilenameProvider(), PurePath(mkdtemp()))

# All Devices created within this block will be
# connected and named at the end of the with block
with init_devices():
    # Create a sample stage with X and Y motors that report their positions
    # to the pattern generator
    stage = sim.SimStage(pattern_generator)
    # Make a detector device that gives the point value of the pattern generator
    # when triggered
    pdet = sim.SimPointDetector(pattern_generator)
    # Make a detector device that gives a gaussian blob with intensity based
    # on the point value of the pattern generator when triggered
    bdet = sim.SimBlobDetector(path_provider, pattern_generator)

```

--------------------------------

### Initialize Devices in Sync Context Manager with RunEngine

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/device-connection-strategies.md

When using a RunEngine (RE fixture), `init_devices()` can be used as a synchronous context manager. It leverages the RunEngine's background event loop for device connections.

```python
import pytest

@pytest.fixture
def my_device(RE):
    with init_devices():
        device = MyDevice()
    return device
```

--------------------------------

### Old TriggerInfo

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/decisions/0012-detector-rewrite.md

Example of the old TriggerInfo structure.

```python
# old
TriggerInfo(
    number_of_events=10,
    trigger=DetectorTrigger.EDGE_TRIGGER,
    livetime=0.1,
    deadtime=0.01,
)
```

--------------------------------

### Create and Activate Virtual Environment

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/tutorials/installation.md

Set up an isolated Python environment to manage project dependencies.

```bash
$ python3 -m venv /path/to/venv
$ source /path/to/venv/bin/activate
```

--------------------------------

### New TriggerInfo

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/decisions/0012-detector-rewrite.md

Example of the updated TriggerInfo structure with EXTERNAL_EDGE trigger.

```python
# new  
TriggerInfo(
    number_of_events=10,
    trigger=DetectorTrigger.EXTERNAL_EDGE,
    livetime=0.1,
    deadtime=0.01,
)
```

--------------------------------

### Load Device Settings with YamlSettingsProvider and apply_panda_settings

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/how-to/store-and-retrieve.md

Load device settings from a YAML file using `YamlSettingsProvider` and `retrieve_settings`, then apply them using `apply_panda_settings` for devices like PandA that require specific ordering.

```python
def load_panda(panda1: HDFPanda):
    provider = YamlSettingsProvider("directory_to_save_yaml_to")
    settings = yield from retrieve_settings(provider, "yaml_file_name", panda1)
    yield from apply_panda_settings(settings)
```

--------------------------------

### Get Single Signal Value

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/how-to/interact-with-signals.md

Retrieve the current value of a single signal. This is an asynchronous operation.

```python
value = await signal.get_value()
```

--------------------------------

### Initialize Devices in Async Context Manager

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/device-connection-strategies.md

Use `init_devices()` as an async context manager for creating multiple devices in parallel within an async test or fixture. This also runs tasks in the current event loop.

```python
import pytest

@pytest.fixture
async def my_device():
    async with init_devices():
        device = MyDevice()
    return device
```

--------------------------------

### Procedural DeviceVector Definition (Ophyd-Async)

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/decisions/0006-procedural-device-definitions.md

Example of defining a group of sensors using DeviceVector in a procedural manner.

```python
class SensorGroup(Device):
    def __init__(self, prefix: str, num: int, name: Optional[str]=None):
        self.sensors = DeviceVector(
            {i: Sensor(f"{prefix}:CHAN{i}" for i in range(1, num+1))}
        )
        super().__init__(name=name)
```

--------------------------------

### Declarative Device Definition (Ophyd)

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/decisions/0006-procedural-device-definitions.md

Example of a traditional declarative device definition using Components in Ophyd.

```python
class Sensor(Device):
    mode = Component(EpicsSignal, "Mode", kind="config")
    value = Component(EpicsSignalRO, "Value", kind="hinted")
```

--------------------------------

### Get Multiple Signal Values Concurrently

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/how-to/interact-with-signals.md

Efficiently retrieve values from multiple signals simultaneously using `asyncio.gather`.

```python
value1, value2 = await asyncio.gather(signal1.get_value(), signal2.get_value())
```

--------------------------------

### Initializing the RunEngine

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/tutorials/using-devices.md

Creates a RunEngine instance with `call_returns_result=True` for capturing results and `autoawait_in_bluesky_event_loop()` to manage asynchronous operations within the same event loop.

```python
RE = RunEngine(call_returns_result=True)
autoawait_in_bluesky_event_loop()

```

--------------------------------

### Get Ophyd-Async Device Name

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/tutorials/using-devices.md

Demonstrates how to access the name attribute of an ophyd-async device. Device names are typically assigned during initialization.

```python
stage.x.name
```

--------------------------------

### Using a Helper Function for Signal Creation

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/devices-signals-backends.md

Demonstrates a shorter way to create a read-only signal using a helper function, abstracting away some of the verbosity of direct backend instantiation.

```python
my_signal = my_cs_signal_r(int, "something")
```

--------------------------------

### Creating and Connecting Devices with init_devices Context Manager

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/tutorials/using-devices.md

Initializes and connects simulated stage and detector devices within the `init_devices` context manager. This context handles device naming, parallel connection, and optional mocking.

```python
with init_devices():
    # Create a sample stage with X and Y motors that report their positions
    # to the pattern generator
    stage = sim.SimStage(pattern_generator)
    # Make a detector device that gives the point value of the pattern generator
    # when triggered
    pdet = sim.SimPointDetector(pattern_generator)
    # Make a detector device that gives a gaussian blob with intensity based
    # on the point value of the pattern generator when triggered
    bdet = sim.SimBlobDetector(path_provider, pattern_generator)

```

--------------------------------

### Define an AreaDetector Subclass

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/how-to/implement-ad-detector.md

Example of creating an AravisDetector subclass, initializing the driver IO and calling the base class constructor with necessary logic and parameters.

```python
class AravisDetector(AreaDetector[AravisDriverIO]):
    """Create an ADAravis AreaDetector instance.

    :param prefix: EPICS PV prefix for the detector
    :param writer_factories: Factories for file writer plugins and their data logics
    :param driver_suffix: Suffix for the driver PV, defaults to "cam1:"
    :param override_deadtime:
        If provided, this value is used for deadtime instead of looking up
        based on camera model.
    :param plugins: Additional areaDetector plugins to include
    :param config_sigs: Additional signals to include in configuration
    :param name: Name for the detector device
    """

    def __init__(
        self,
        prefix: str,
        *writer_factories: ADWriterFactory,
        driver_suffix="cam1:",
        override_deadtime: float | None = None,
        plugins: dict[str, NDPluginBaseIO] | None = None,
        config_sigs: Sequence[SignalR] = (),
        name: str = "",
    ) -> None:
        driver = AravisDriverIO(prefix + driver_suffix)
        super().__init__(
            driver,
            prefix,
            *writer_factories,
            acquire_logic=ADAcquireLogic(driver),
            trigger_logic=AravisTriggerLogic(driver, override_deadtime),
            plugins=plugins,
            config_sigs=config_sigs,
            name=name,
        )
```

--------------------------------

### Create Mock Motor Fixture

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/tutorials/writing-tests-for-devices.md

Sets up a mock DemoMotor using init_devices and configures its signals with mock values. This fixture is asynchronous and intended for use with pytest-asyncio.

```python
import pytest
import ophyd.sim as sim
from ophyd.utils import setup_device
from ophyd.device import Device

# Assuming demo and init_devices are available from context
# from . import demo
# from .utils import init_devices, set_mock_value, assert_reading, assert_configuration, assert_value


@pytest.fixture
async def mock_motor():
    async with init_devices(mock=True):
        mock_motor = demo.DemoMotor("BLxxI-MO-TABLE-01:X:")
    set_mock_value(mock_motor.units, "mm")
    set_mock_value(mock_motor.precision, 3)
    set_mock_value(mock_motor.velocity, 1)
    yield mock_motor
```

--------------------------------

### Add Stats Plugin Data Logic

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/how-to/implement-ad-detector.md

Example of adding a stats plugin's total signal as a readable signal in events alongside file writing.

```python
from ophyd_async.epics.adcore import PluginSignalDataLogic

det = adaravis.AravisDetector(prefix, path_provider)
# Add stats total as a readable signal in events
det.add_detector_logics(adcore.PluginSignalDataLogic(det.driver, det.stats.total))
```

--------------------------------

### Detector plan with prepared trigger information

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/tutorials/implementing-detectors.md

Prepares the detector with trigger information before the first trigger call, optimizing scan startup.

```python
from bluesky.plan_stubs import trigger, read
from bluesky.preprocessors import baseline_prepare
from ophyd_async.core import TriggerInfo

@baseline_prepare
def prepared_detector_plan(detector):
    # Prepare detector with specific trigger parameters
    yield from detector.prepare(trigger_info=TriggerInfo(livetime=0.2, deadtime=0.1))
    yield from trigger(detector)
    yield from read(detector)

# Usage:
# RE(prepared_detector_plan(my_detector))

```

--------------------------------

### Procedural Device Definition with Context Manager (Ophyd-Async)

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/decisions/0006-procedural-device-definitions.md

A potential future improvement for procedural device definitions using context managers to reduce verbosity.

```python
with self.signals_added_to(READ):
    self.value = epics_signal_r(float, prefix + "Value")
with self.signals_added_to(CONFIG):
    self.mode = epics_signal_rw(EnergyMode, prefix + "Mode")
```

--------------------------------

### ADWriterFactory HDF with Callable Override

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/decisions/0016-ad-writer-factory.md

Example of using ADWriterFactory.hdf with a callable to provide a custom NDArrayDescription, useful when signals come from a different source like an ROI plugin.

```python
ADWriterFactory.hdf(
    path_provider,
    writer_name="hdf1",
    datakey_suffix="-roi1",
    array_description=lambda driver: NDArrayDescription(
        shape_signals=(roi1.size_y, roi1.size_x),
        data_type_signal=driver.data_type,   # driver available here
        color_mode_signal=driver.color_mode,
    ),
)
```

--------------------------------

### Declarative Device Definition with Type Hints (Ophyd-Async)

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/decisions/0006-procedural-device-definitions.md

An alternative declarative approach for ophyd-async using type hints.

```python
from typing import Annotated as A

class Sensor(EpicsDevice):
    mode: A[SignalRW, CONFIG, pv_suffix("Mode")]
    value: A[SignalR, READ, pv_suffix("Value")]
```

--------------------------------

### Old SimDetector Initialization

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/decisions/0012-detector-rewrite.md

Initialization of SimDetector using the old controller and writer_cls.with_io pattern.

```python
# old - controller and writer_cls.with_io

from ophyd_async.epics import adcore

class SimDetector(adcore.AreaDetector[SimController]):
    def __init__(
        self, 
        prefix: str,
        path_provider: PathProvider,
        drv_suffix="cam1:",
        writer_cls: type[adcore.ADWriter] = adcore.ADHDFWriter,
        fileio_suffix: str | None = None,
        name="",
        config_sigs: Sequence[SignalR] = (),
        plugins: dict[str, adcore.NDPluginBaseIO] | None = None,
    ):
        driver = adcore.ADBaseIO(prefix + drv_suffix)
        controller = SimController(driver)
        writer = writer_cls.with_io(
            prefix,
            path_provider,
            dataset_source=driver,
            fileio_suffix=fileio_suffix,
            plugins=plugins,
        )
        super().__init__(
            controller=controller,
            writer=writer,
            plugins=plugins,
            name=name,
            config_sigs=config_sigs,
        )
```

--------------------------------

### Load Device Settings Conditionally with apply_settings_if_different

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/how-to/store-and-retrieve.md

Use `apply_settings_if_different` to load settings only if the current device values differ from the stored ones. This can be more efficient when reading signals is faster than setting them. It requires a `SettingsProvider`, a file name, the device, and an optional plan for applying settings.

```python
def load_panda(panda1: HDFPanda):
    provider = YamlSettingsProvider("directory_to_save_yaml_to")
    settings = yield from retrieve_settings(provider, "yaml_file_name", panda1)
    yield from apply_settings_if_different(settings, apply_panda_settings)
```

--------------------------------

### Read-Only Derived Signal Example

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/how-to/derive-one-signal-from-others.md

Demonstrates a read-only derived signal that calculates its position based on the raw x and y motor signals. Use this when you only need to read a value derived from other signals.

```python
from __future__ import annotations

import asyncio

from ophyd_async.core import (
    Device,
    DeviceVector,
    StandardReadable,
    StrictEnum,
    derived_signal_r,
    derived_signal_rw,
    derived_signal_w,
    soft_signal_rw,
)


class BeamstopPosition(StrictEnum):
    IN_POSITION = "In position"
    OUT_OF_POSITION = "Out of position"


class ReadOnlyBeamstop(Device):
    """Reads from 2 motors to work out if the beamstop is in position.

    E.g. bps.rd(beamstop.position)
    """

    def __init__(self, name=""):
        # Raw signals
        self.x = soft_signal_rw(float)
        self.y = soft_signal_rw(float)
        # Derived signals
        self.position = derived_signal_r(self._get_position, x=self.x, y=self.y)
        super().__init__(name=name)

    def _get_position(self, x: float, y: float) -> BeamstopPosition:
        if abs(x) < 1 and abs(y) < 2:
            return BeamstopPosition.IN_POSITION
        else:
            return BeamstopPosition.OUT_OF_POSITION

```

--------------------------------

### Threaded Signal Setting (Ophyd Sync)

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/design-goals.md

Illustrates the traditional threaded approach for setting multiple signals in ophyd sync. This method involves creating and managing OS threads, which can introduce overhead and complexity for coordination.

```python
def set_signal_thread(signal, value):
    t = Thread(signal.set, value)
    t.start()
    return

def run():
    t1 = set_signal_thread(signal1, value1)
    t2 = set_signal_thread(signal2, value2)
    t1.join()
    t2.join()
    value = signal3.get_value()
```

--------------------------------

### Tango Procedural Device Definition

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/decisions/0009-procedural-vs-declarative-devices.md

Defines a Tango device using a procedural approach, similar to the EPICS procedural example. This demonstrates how to instantiate Tango signals within an __init__ method.

```python
class TangoProceduralDevice(StandardReadable):
    def __init__(self, prefix: str, name="") -> None:
        with self.add_children_as_readables():
            self.value = DeviceVector({0: tango_signal_r(float)})
        with self.add_children_as_readables(ConfigSignal):
            self.mode = tango_signal_rw(EnergyMode)
        super().__init__(name=name, connector=TangoConnector(prefix))
```

--------------------------------

### Move Device to Position using `bps.mv` Plan

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/tutorials/using-devices.md

Executes a bluesky plan stub (`bps.mv`) to move a device to a specified position. The RunEngine executes the plan and returns the status of the move operation.

```python
RE(bps.mv(stage.x, 1.5))
```

--------------------------------

### Configuring Static Path Provider for File Writing

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/tutorials/using-devices.md

Sets up a `StaticPathProvider` with a `UUIDFilenameProvider` to specify a temporary directory for file-writing detectors and name files using UUIDs.

```python
path_provider = StaticPathProvider(UUIDFilenameProvider(), PurePath(mkdtemp()))

```

--------------------------------

### Test a Device within a Bluesky Plan

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/tutorials/writing-tests-for-devices.md

Execute a bluesky plan with a mocked Device and assert the produced documents and data. This verifies Device behavior within a larger experimental workflow.

```python
@pytest.fixture
async def mock_point_detector():
    async with init_devices(mock=True):
        mock_point_detector = demo.DemoPointDetector("MOCK:DET:")
    yield mock_point_detector

async def test_point_detector_in_plan(
    RE: RunEngine, mock_point_detector: demo.DemoPointDetector
):
    # Subscribe to new documents produce, putting them in a dict by type
    docs = defaultdict(list)
    RE.subscribe(lambda name, doc: docs[name].append(doc))
    # Set the channel values to a known value
    for i, channel in mock_point_detector.channel.items():
        set_mock_value(channel.value, 100 + i)
    # Run the plan and assert the right docs are produced
    RE(bp.count([mock_point_detector], num=2))
    assert_emitted(docs, start=1, descriptor=1, event=2, stop=1)
    assert docs["event"][1]["data"] == {
        "mock_point_detector-channel-1-value": 101,
        "mock_point_detector-channel-2-value": 102,
        "mock_point_detector-channel-3-value": 103,
    }
```

--------------------------------

### Connect Device using ensure_connected Plan

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/device-connection-strategies.md

Use the `ensure_connected` plan to connect a Device when a RunEngine is active. This is often used directly in a test case rather than a fixture.

```python
import pytest

@pytest.fixture
def my_device(RE):
    device = MyDevice(name="device")
    RE(ensure_connected(device))
    return device
```

--------------------------------

### Google Style Docstring with RST Links

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/decisions/0010-docstring-format.md

Illustrates the Google docstring style with RST links for arguments and yields. This style is often more concise than Numpy style.

```python
    def create_devices_from_annotations(
        self,
        filled=True,
    ) -> Iterator[tuple[DeviceConnectorT, list[Any]]]:
        """Create all Signals from annotations

        Args:
            filled: If ``True`` then the Devices created should be considered
                already filled with connection data. If ``False`` then
                `fill_child_device` needs calling at parent device connection
                time before the child `Device` can be connected.

        Yields:
            (connector, extras): The `DeviceConnector` that has been created for this
                Signal, and the list of extra annotations that could be used to
                customize it.
        """

```

--------------------------------

### Multiple Data Streams (New)

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/decisions/0012-detector-rewrite.md

Demonstrates adding multiple ADHDFDataLogic instances to handle different data streams, such as separate HDF writers for different ROIs.

```python
# new - add multiple data logics
detector = AreaDetector(
    driver=driver,
    arm_logic=ADArmLogic(driver),
    writer_type=None,  # Don't create default writer
)
# Add separate HDF writers for different ROIs
detector.add_detector_logics(
    ADHDFDataLogic(..., datakey_suffix="-roi1"),
    ADHDFDataLogic(..., datakey_suffix="-roi2"),
)
```

--------------------------------

### Procedural Device Creation with Signals

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/declarative-vs-procedural.md

Defines device signals within an __init__ method, offering explicit control and flexibility. Use when embedding arbitrary Python logic or creating complex soft signals.

```default
    """For usage when simulating a motor."""

    def __init__(
        self,
        name: str = "",
        instant: bool = True,
        initial_value: float = 0.0,
        units: str = "mm",
    ) -> None:
        """Simulate a motor, with optional velocity.

        :param name: name of device
        :param instant: whether to move instantly or calculate move time using velocity
        :param initial_value: initial position of the motor
        :param units: units of the motor position
        """
        # Define some signals
        with self.add_children_as_readables(Format.HINTED_SIGNAL):
            self.user_readback, self._user_readback_set = soft_signal_r_and_setter(
                float, 0
            )
        with self.add_children_as_readables(Format.CONFIG_SIGNAL):
            self.velocity = soft_signal_rw(float, 0 if instant else 1.0)
            self.acceleration_time = soft_signal_rw(float, 0.5)
            self.units = soft_signal_rw(str, units)
        self.user_setpoint = soft_signal_rw(float, initial_value)

        # Whether set() should complete successfully or not
        self._set_success = True
        self._move_status: AsyncStatus | None = None
        # Stored in prepare
        self._fly_info: FlyMotorInfo | None = None
        # Set on kickoff(), complete when motor reaches end position
        self._fly_status: WatchableAsyncStatus | None = None

        super().__init__(name=name)

```

--------------------------------

### Configuring Multiple HDF Writers with ADWriterFactory

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/decisions/0016-ad-writer-factory.md

Demonstrates setting up an Area Detector with multiple HDF writers using ADWriterFactory, each with a unique name and datakey suffix.

```python
det = adaravis.AravisDetector(
    "PREFIX:",
    ADWriterFactory.hdf(path_provider, writer_name="hdf1", datakey_suffix="-roi1",
                        array_description=lambda driver: NDArrayDescription(...)),
    ADWriterFactory.hdf(path_provider, writer_name="hdf2", datakey_suffix="-roi2",
                        array_description=lambda driver: NDArrayDescription(...)),
    plugins={"roi1": roi1, "roi2": roi2},
)
det.hdf1  # NDFileHDF5IO for ROI 1
det.hdf2  # NDFileHDF5IO for ROI 2
```

--------------------------------

### Detector plan with descriptor preparation

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/tutorials/implementing-detectors.md

Moves the creation of the event descriptor to an earlier stage, ensuring no extra work is needed on the first trigger call.

```python
from bluesky.plan_stubs import trigger, read, prepare
from bluesky.preprocessors import baseline_prepare
from ophyd_async.core import TriggerInfo

@baseline_prepare
def early_descriptor_plan(detector):
    # Prepare detector and create descriptor early
    yield from prepare(detector, trigger_info=TriggerInfo(livetime=0.2, deadtime=0.1))
    yield from trigger(detector)
    yield from read(detector)

# Usage:
# RE(early_descriptor_plan(my_detector))

```

--------------------------------

### Use soft_signal_rw for Mock Signals

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/decisions/0008-signal-types.md

The MockSignalBackend is no longer directly supported. Use soft_signal_rw and connect with mock=True to create mock signals. This ensures proper signal lifecycle management.

```python
# old
fake_set_signal = SignalRW(MockSignalBackend(float))
# new
fake_set_signal = soft_signal_rw(float)
await fake_set_signal.connect(mock=True)
```

--------------------------------

### Mock a Device for Plan Execution

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/tutorials/writing-tests-for-devices.md

Use init_devices in mock mode within a pytest fixture to create a mock instance of a Device that will be used in bluesky plans.

```python
@pytest.fixture
async def mock_point_detector():
    async with init_devices(mock=True):
        mock_point_detector = demo.DemoPointDetector("MOCK:DET:")
    yield mock_point_detector
```

--------------------------------

### Detector Controller to Trigger Logic + Arm Logic Migration

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/decisions/0012-detector-rewrite.md

Illustrates the refactoring of detector controller logic into separate Trigger and Arm logic components. This change is necessary for all existing detector implementations.

```python
class NewDetector(StandardDetector):
    def __init__(self, name, **kwargs):
        super().__init__(name, **kwargs)
        self.add_detector_logics(
            DetectorTriggerLogic(self),
            DetectorArmLogic(self),
            DetectorDataLogic(self),
        )
        self.add_config_signals(self.some_config_signal)

    def some_config_signal(self):
        # ...
        pass
```

--------------------------------

### Store Device Settings with YamlSettingsProvider

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/how-to/store-and-retrieve.md

Use `store_settings` with `YamlSettingsProvider` to save the current values of a device's SignalRWs to a YAML file. This requires a connected RunEngine and a device instance.

```python
provider = YamlSettingsProvider("directory_to_save_yaml_to")
RE(store_settings(provider, "yaml_file_name", panda1))
```

--------------------------------

### Multiple Data Streams (Old)

Source: https://github.com/bluesky/ophyd-async/blob/main/docs/explanations/decisions/0012-detector-rewrite.md

Illustrates the old approach where handling multiple data streams required complex inheritance.

```python
# old - required complex inheritance
```