|
| 1 | +from collections.abc import Sequence |
| 2 | +from typing import Annotated, Any |
| 3 | + |
| 4 | +import bluesky.plans as bp |
| 5 | +from bluesky.protocols import Readable |
| 6 | +from bluesky.utils import MsgGenerator |
| 7 | +from ophyd_async.core import AsyncReadable |
| 8 | +from pydantic import Field, NonNegativeFloat, validate_call |
| 9 | + |
| 10 | + |
| 11 | +@validate_call(config={"arbitrary_types_allowed": True}) |
| 12 | +def count( |
| 13 | + detectors: Annotated[ |
| 14 | + Sequence[Readable | AsyncReadable], |
| 15 | + Field( |
| 16 | + description="Set of readable devices, will take a reading at each point", |
| 17 | + min_length=1, |
| 18 | + ), |
| 19 | + ], |
| 20 | + num: Annotated[int, Field(description="Number of frames to collect", ge=1)] = 1, |
| 21 | + delay: Annotated[ |
| 22 | + NonNegativeFloat | Sequence[NonNegativeFloat], |
| 23 | + Field( |
| 24 | + description="Delay between readings: if tuple, len(delay) == num - 1 and \ |
| 25 | + the delays are between each point, if value or None is the delay for every \ |
| 26 | + gap", |
| 27 | + json_schema_extra={"units": "s"}, |
| 28 | + ), |
| 29 | + ] = 0.0, |
| 30 | + metadata: dict[str, Any] | None = None, |
| 31 | +) -> MsgGenerator: |
| 32 | + """Reads from a number of devices. |
| 33 | +
|
| 34 | + Wraps bluesky.plans.count(det, num, delay, md=metadata) exposing only serializable |
| 35 | + parameters and metadata. |
| 36 | + """ |
| 37 | + if isinstance(delay, Sequence): |
| 38 | + assert len(delay) == num - 1, ( |
| 39 | + f"Number of delays given must be {num - 1}: was given {len(delay)}" |
| 40 | + ) |
| 41 | + metadata = metadata or {} |
| 42 | + metadata["shape"] = (num,) |
| 43 | + yield from bp.count(tuple(detectors), num, delay=delay, md=metadata) |
0 commit comments