-
Notifications
You must be signed in to change notification settings - Fork 5
Open
Labels
bugSomething isn't workingSomething isn't working
Description
In the following test (apologies for non-minimal example), I have to check that all of my expected dimensions are in the dimensions, rather than checking that the dimensions is equal to a list. As dimensions is a list, its order should be deterministic.
Additionally, the dimensions are a list[tuple(list[str], str)], not a list[list[str | list[str]] as expected. Probably this is an error with event-model and the correct typing is list[tuple[list[str], str]].
# @attach_data_session_metadata_decorator()
@validate_call(config={"arbitrary_types_allowed": True})
def spec_scan(
detectors: Annotated[
set[Readable],
Field(
description="Set of readable devices, will take a reading at each point, \
in addition to any Movables in the Spec",
),
],
spec: Annotated[
Spec[Movable],
Field(description="ScanSpec modelling the path of the scan"),
],
metadata: dict[str, Any] | None = None,
) -> MsgGenerator:
"""Generic plan for reading `detectors` at every point of a ScanSpec `Spec`.
A `Spec` is an N-dimensional path.
"""
_md = {
"plan_args": {
"detectors": {det.name for det in detectors},
"spec": repr(spec),
},
"plan_name": "spec_scan",
"shape": spec.shape(),
**(metadata or {}),
}
yield from bp.scan_nd(tuple(detectors), _as_cycler(spec), md=_md)
def _as_cycler(spec: Spec[Movable]) -> Cycler:
"""
Convert a scanspec to a cycler for compatibility with legacy Bluesky plans such as
`bp.scan_nd`. Use the midpoints of the scanspec since cyclers are normally used
for software triggered scans.
Args:
spec: A scanspec
Returns:
Cycler: A new cycler
"""
midpoints = spec.frames().midpoints
# Need to "add" the cyclers for all the axes together. The code below is
# effectively: cycler(motor1, [...]) + cycler(motor2, [...]) + ...
return reduce(operator.add, (cycler(*args) for args in midpoints.items()))
@pytest.fixture
def documents_from_expected_shape(
request: pytest.FixtureRequest,
det: StandardDetector,
path_provider,
RE: RunEngine,
x_axis: SimMotor,
y_axis: SimMotor,
) -> dict[str, list[DocumentType]]:
shape: Sequence[int] = request.param
motors = [x_axis, y_axis]
# spec = Static.duration(1)
spec = Line(motors[0], 0, 5, shape[0])
for i in range(1, len(shape)):
spec = spec * Line(motors[i], 0, 5, shape[i])
docs: dict[str, list[DocumentType]] = {}
RE(
spec_scan({det}, spec), # type: ignore
lambda name, doc: docs.setdefault(name, []).append(doc),
)
return docs
spec_and_shape = (
# [(), (1,)], # static
[(1,), (1,)],
[(3,), (3,)],
[(1, 1), (1, 1)],
[(3, 3), (3, 3)],
)
@pytest.mark.parametrize(
"documents_from_expected_shape, shape",
spec_and_shape,
indirect=["documents_from_expected_shape"],
)
def test_plan_produces_expected_start_document(
documents_from_expected_shape: dict[str, list[DocumentType]],
shape: tuple[int, ...],
x_axis: SimMotor,
y_axis: SimMotor,
):
axes = len(shape)
expected_data_keys = (
[
x_axis.hints.get("fields", [])[0],
y_axis.hints.get("fields", [])[0],
]
if axes == 2
else [x_axis.hints.get("fields", [])[0]]
)
dimensions = [([data_key], "primary") for data_key in expected_data_keys]
docs = documents_from_expected_shape.get("start")
assert docs and len(docs) == 1
start = cast(RunStart, docs[0])
assert start.get("shape") == shape
assert (hints := start.get("hints"))
for dimension in dimensions:
assert dimension in hints.get("dimensions") # type: ignore
Metadata
Metadata
Assignees
Labels
bugSomething isn't workingSomething isn't working