import sys
import inspect
from itertools import chain, zip_longest
from functools import partial
import collections
from collections import defaultdict
import time
import numpy as np
try:
# cytools is a drop-in replacement for toolz, implemented in Cython
from cytools import partition
except ImportError:
from toolz import partition
from . import plan_patterns
from . import utils
from .utils import Msg
from . import preprocessors as bpp
from . import plan_stubs as bps
[docs]def count(detectors, num=1, delay=None, *, per_shot=None, md=None):
"""
Take one or more readings from detectors.
Parameters
----------
detectors : list
list of 'readable' objects
num : integer, optional
number of readings to take; default is 1
If None, capture data until canceled
delay : iterable or scalar, optional
Time delay in seconds between successive readings; default is 0.
per_shot : callable, optional
hook for customizing action of inner loop (messages per step)
Expected signature ::
def f(detectors: Iterable[OphydObj]) -> Generator[Msg]:
...
md : dict, optional
metadata
Notes
-----
If ``delay`` is an iterable, it must have at least ``num - 1`` entries or
the plan will raise a ``ValueError`` during iteration.
"""
if num is None:
num_intervals = None
else:
num_intervals = num - 1
_md = {'detectors': [det.name for det in detectors],
'num_points': num,
'num_intervals': num_intervals,
'plan_args': {'detectors': list(map(repr, detectors)), 'num': num},
'plan_name': 'count',
'hints': {}
}
_md.update(md or {})
_md['hints'].setdefault('dimensions', [(('time',), 'primary')])
if per_shot is None:
per_shot = bps.one_shot
@bpp.stage_decorator(detectors)
@bpp.run_decorator(md=_md)
def inner_count():
return (yield from bps.repeat(partial(per_shot, detectors),
num=num, delay=delay))
return (yield from inner_count())
[docs]def list_scan(detectors, *args, per_step=None, md=None):
"""
Scan over one or more variables in steps simultaneously (inner product).
Parameters
----------
detectors : list
list of 'readable' objects
*args :
For one dimension, ``motor, [point1, point2, ....]``.
In general:
.. code-block:: python
motor1, [point1, point2, ...],
motor2, [point1, point2, ...],
...,
motorN, [point1, point2, ...]
Motors can be any 'settable' object (motor, temp controller, etc.)
per_step : callable, optional
hook for customizing action of inner loop (messages per step)
Expected signature:
``f(detectors, motor, step) -> plan (a generator)``
md : dict, optional
metadata
See Also
--------
:func:`bluesky.plans.rel_list_scan`
:func:`bluesky.plans.list_grid_scan`
:func:`bluesky.plans.rel_list_grid_scan`
"""
if len(args) % 2 != 0:
raise ValueError("The list of arguments must contain a list of "
"points for each defined motor")
md = md or {} # reset md if it is None.
# set some variables and check that all lists are the same length
lengths = {}
motors = []
pos_lists = []
length = None
for motor, pos_list in partition(2, args):
pos_list = list(pos_list) # Ensure list (accepts any finite iterable).
lengths[motor.name] = len(pos_list)
if not length:
length = len(pos_list)
motors.append(motor)
pos_lists.append(pos_list)
length_check = all(elem == list(lengths.values())[0] for elem in
list(lengths.values()))
if not length_check:
raise ValueError("The lengths of all lists in *args must be the same. "
"However the lengths in args are : "
"{}".format(lengths))
md_args = list(chain(*((repr(motor), pos_list)
for motor, pos_list in partition(2, args))))
motor_names = list(lengths.keys())
_md = {'detectors': [det.name for det in detectors],
'motors': motor_names,
'num_points': length,
'num_intervals': length - 1,
'plan_args': {'detectors': list(map(repr, detectors)),
'args': md_args,
'per_step': repr(per_step)},
'plan_name': 'list_scan',
'plan_pattern': 'inner_list_product',
'plan_pattern_module': plan_patterns.__name__,
'plan_pattern_args': dict(args=md_args),
'hints': {},
}
_md.update(md or {})
x_fields = []
for motor in motors:
x_fields.extend(getattr(motor, 'hints', {}).get('fields', []))
default_dimensions = [(x_fields, 'primary')]
default_hints = {}
if len(x_fields) > 0:
default_hints.update(dimensions=default_dimensions)
# now add default_hints and override any hints from the original md (if
# exists)
_md['hints'] = default_hints
_md['hints'].update(md.get('hints', {}) or {})
full_cycler = plan_patterns.inner_list_product(args)
return (yield from scan_nd(detectors, full_cycler, per_step=per_step,
md=_md))
[docs]def rel_list_scan(detectors, *args, per_step=None, md=None):
"""
Scan over one variable in steps relative to current position.
Parameters
----------
detectors : list
list of 'readable' objects
*args :
For one dimension, ``motor, [point1, point2, ....]``.
In general:
.. code-block:: python
motor1, [point1, point2, ...],
motor2, [point1, point2, ...],
...,
motorN, [point1, point2, ...]
Motors can be any 'settable' object (motor, temp controller, etc.)
point1, point2 etc are relative to the current location.
motor : object
any 'settable' object (motor, temp controller, etc.)
steps : list
list of positions relative to current position
per_step : callable, optional
hook for customizing action of inner loop (messages per step)
Expected signature: ``f(detectors, motor, step)``
md : dict, optional
metadata
See Also
--------
:func:`bluesky.plans.list_scan`
:func:`bluesky.plans.list_grid_scan`
:func:`bluesky.plans.rel_list_grid_scan`
"""
# TODO read initial positions (redundantly) so they can be put in md here
_md = {'plan_name': 'rel_list_scan'}
_md.update(md or {})
motors = [motor for motor, pos_list in partition(2, args)]
@bpp.reset_positions_decorator(motors)
@bpp.relative_set_decorator(motors)
def inner_relative_list_scan():
return (yield from list_scan(detectors, *args, per_step=per_step,
md=_md))
return (yield from inner_relative_list_scan())
[docs]def list_grid_scan(detectors, *args, snake_axes=False, per_step=None, md=None):
"""
Scan over a mesh; each motor is on an independent trajectory.
Parameters
----------
detectors: list
list of 'readable' objects
args: list
patterned like (``motor1, position_list1,``
``motor2, position_list2,``
``motor3, position_list3,``
``...,``
``motorN, position_listN``)
The first motor is the "slowest", the outer loop. ``position_list``'s
are lists of positions, all lists must have the same length. Motors
can be any 'settable' object (motor, temp controller, etc.).
snake_axes: boolean or iterable, optional
which axes should be snaked, either ``False`` (do not snake any axes),
``True`` (snake all axes) or a list of axes to snake. "Snaking" an axis
is defined as following snake-like, winding trajectory instead of a
simple left-to-right trajectory.The elements of the list are motors
that are listed in `args`. The list must not contain the slowest
(first) motor, since it can't be snaked.
per_step: callable, optional
hook for customizing action of inner loop (messages per step).
See docstring of :func:`bluesky.plan_stubs.one_nd_step` (the default)
for details.
md: dict, optional
metadata
See Also
--------
:func:`bluesky.plans.rel_list_grid_scan`
:func:`bluesky.plans.list_scan`
:func:`bluesky.plans.rel_list_scan`
"""
full_cycler = plan_patterns.outer_list_product(args, snake_axes)
md_args = []
motor_names = []
motors = []
for i, (motor, pos_list) in enumerate(partition(2, args)):
md_args.extend([repr(motor), pos_list])
motor_names.append(motor.name)
motors.append(motor)
_md = {'shape': tuple(len(pos_list)
for motor, pos_list in partition(2, args)),
'extents': tuple([min(pos_list), max(pos_list)]
for motor, pos_list in partition(2, args)),
'snake_axes': snake_axes,
'plan_args': {'detectors': list(map(repr, detectors)),
'args': md_args,
'per_step': repr(per_step)},
'plan_name': 'list_grid_scan',
'plan_pattern': 'outer_list_product',
'plan_pattern_args': dict(args=md_args).update(
{'snake_axes': snake_axes}),
'plan_pattern_module': plan_patterns.__name__,
'motors': tuple(motor_names),
'hints': {},
}
_md.update(md or {})
try:
_md['hints'].setdefault('dimensions', [(m.hints['fields'], 'primary')
for m in motors])
except (AttributeError, KeyError):
...
return (yield from scan_nd(detectors, full_cycler,
per_step=per_step, md=_md))
[docs]def rel_list_grid_scan(detectors, *args, snake_axes=False, per_step=None,
md=None):
"""
Scan over a mesh; each motor is on an independent trajectory. Each point is
relative to the current position.
Parameters
----------
detectors : list
list of 'readable' objects
args
patterned like (``motor1, position_list1,``
``motor2, position_list2,``
``motor3, position_list3,``
``...,``
``motorN, position_listN``)
The first motor is the "slowest", the outer loop. ``position_list``'s
are lists of positions, all lists must have the same length. Motors
can be any 'settable' object (motor, temp controller, etc.).
snake_axes : boolean or Iterable, optional
which axes should be snaked, either ``False`` (do not snake any axes),
``True`` (snake all axes) or a list of axes to snake. "Snaking" an axis
is defined as following snake-like, winding trajectory instead of a
simple left-to-right trajectory.The elements of the list are motors
that are listed in `args`. The list must not contain the slowest
(first) motor, since it can't be snaked.
per_step : callable, optional
hook for customizing action of inner loop (messages per step).
See docstring of :func:`bluesky.plan_stubs.one_nd_step` (the default)
for details.
md : dict, optional
metadata
See Also
--------
:func:`bluesky.plans.list_grid_scan`
:func:`bluesky.plans.list_scan`
:func:`bluesky.plans.rel_list_scan`
"""
_md = {'plan_name': 'rel_list_grid_scan'}
_md.update(md or {})
motors = [motor for motor, pos_list in partition(2, args)]
@bpp.reset_positions_decorator(motors)
@bpp.relative_set_decorator(motors)
def inner_relative_list_grid_scan():
return (yield from list_grid_scan(detectors, *args,
snake_axes=snake_axes,
per_step=per_step, md=_md))
return (yield from inner_relative_list_grid_scan())
def _scan_1d(detectors, motor, start, stop, num, *, per_step=None, md=None):
"""
Scan over one variable in equally spaced steps.
Parameters
----------
detectors : list
list of 'readable' objects
motor : object
any 'settable' object (motor, temp controller, etc.)
start : float
starting position of motor
stop : float
ending position of motor
num : int
number of steps
per_step : callable, optional
hook for customizing action of inner loop (messages per step)
Expected signature: ``f(detectors, motor, step)``
md : dict, optional
metadata
See Also
--------
:func:`bluesky.plans.rel_scan`
"""
_md = {'detectors': [det.name for det in detectors],
'motors': [motor.name],
'num_points': num,
'num_intervals': num - 1,
'plan_args': {'detectors': list(map(repr, detectors)), 'num': num,
'motor': repr(motor),
'start': start, 'stop': stop,
'per_step': repr(per_step)},
'plan_name': 'scan',
'plan_pattern': 'linspace',
'plan_pattern_module': 'numpy',
'plan_pattern_args': dict(start=start, stop=stop, num=num),
'hints': {},
}
_md.update(md or {})
try:
dimensions = [(motor.hints['fields'], 'primary')]
except (AttributeError, KeyError):
pass
else:
_md['hints'].setdefault('dimensions', dimensions)
if per_step is None:
per_step = bps.one_1d_step
steps = np.linspace(**_md['plan_pattern_args'])
@bpp.stage_decorator(list(detectors) + [motor])
@bpp.run_decorator(md=_md)
def inner_scan():
for step in steps:
yield from per_step(detectors, motor, step)
return (yield from inner_scan())
def _rel_scan_1d(detectors, motor, start, stop, num, *, per_step=None,
md=None):
"""
Scan over one variable in equally spaced steps relative to current positon.
Parameters
----------
detectors : list
list of 'readable' objects
motor : object
any 'settable' object (motor, temp controller, etc.)
start : float
starting position of motor
stop : float
ending position of motor
num : int
number of steps
per_step : callable, optional
hook for customizing action of inner loop (messages per step)
Expected signature: ``f(detectors, motor, step)``
md : dict, optional
metadata
See Also
--------
:func:`bluesky.plans.scan`
"""
_md = {'plan_name': 'rel_scan'}
_md.update(md or {})
# TODO read initial positions (redundantly) so they can be put in md here
@bpp.reset_positions_decorator([motor])
@bpp.relative_set_decorator([motor])
def inner_relative_scan():
return (yield from _scan_1d(detectors, motor, start, stop,
num, per_step=per_step, md=_md))
return (yield from inner_relative_scan())
[docs]def log_scan(detectors, motor, start, stop, num, *, per_step=None, md=None):
"""
Scan over one variable in log-spaced steps.
Parameters
----------
detectors : list
list of 'readable' objects
motor : object
any 'settable' object (motor, temp controller, etc.)
start : float
starting position of motor
stop : float
ending position of motor
num : int
number of steps
per_step : callable, optional
hook for customizing action of inner loop (messages per step)
Expected signature: ``f(detectors, motor, step)``
md : dict, optional
metadata
See Also
--------
:func:`bluesky.plans.rel_log_scan`
"""
_md = {'detectors': [det.name for det in detectors],
'motors': [motor.name],
'num_points': num,
'num_intervals': num - 1,
'plan_args': {'detectors': list(map(repr, detectors)), 'num': num,
'start': start, 'stop': stop, 'motor': repr(motor),
'per_step': repr(per_step)},
'plan_name': 'log_scan',
'plan_pattern': 'logspace',
'plan_pattern_module': 'numpy',
'plan_pattern_args': dict(start=start, stop=stop, num=num),
'hints': {},
}
_md.update(md or {})
try:
dimensions = [(motor.hints['fields'], 'primary')]
except (AttributeError, KeyError):
pass
else:
_md['hints'].setdefault('dimensions', dimensions)
if per_step is None:
per_step = bps.one_1d_step
steps = np.logspace(**_md['plan_pattern_args'])
@bpp.stage_decorator(list(detectors) + [motor])
@bpp.run_decorator(md=_md)
def inner_log_scan():
for step in steps:
yield from per_step(detectors, motor, step)
return (yield from inner_log_scan())
[docs]def rel_log_scan(detectors, motor, start, stop, num, *, per_step=None,
md=None):
"""
Scan over one variable in log-spaced steps relative to current position.
Parameters
----------
detectors : list
list of 'readable' objects
motor : object
any 'settable' object (motor, temp controller, etc.)
start : float
starting position of motor
stop : float
ending position of motor
num : int
number of steps
per_step : callable, optional
hook for customizing action of inner loop (messages per step)
Expected signature: ``f(detectors, motor, step)``
md : dict, optional
metadata
See Also
--------
:func:`bluesky.plans.log_scan`
"""
# TODO read initial positions (redundantly) so they can be put in md here
_md = {'plan_name': 'rel_log_scan'}
_md.update(md or {})
@bpp.reset_positions_decorator([motor])
@bpp.relative_set_decorator([motor])
def inner_relative_log_scan():
return (yield from log_scan(detectors, motor, start, stop, num,
per_step=per_step, md=_md))
return (yield from inner_relative_log_scan())
[docs]def adaptive_scan(detectors, target_field, motor, start, stop,
min_step, max_step, target_delta, backstep,
threshold=0.8, *, md=None):
"""
Scan over one variable with adaptively tuned step size.
Parameters
----------
detectors : list
list of 'readable' objects
target_field : string
data field whose output is the focus of the adaptive tuning
motor : object
any 'settable' object (motor, temp controller, etc.)
start : float
starting position of motor
stop : float
ending position of motor
min_step : float
smallest step for fast-changing regions
max_step : float
largest step for slow-chaning regions
target_delta : float
desired fractional change in detector signal between steps
backstep : bool
whether backward steps are allowed -- this is concern with some motors
threshold : float, optional
threshold for going backward and rescanning a region, default is 0.8
md : dict, optional
metadata
See Also
--------
:func:`bluesky.plans.rel_adaptive_scan`
"""
if not 0 < min_step < max_step:
raise ValueError("min_step and max_step must meet condition of "
"max_step > min_step > 0")
_md = {'detectors': [det.name for det in detectors],
'motors': [motor.name],
'plan_args': {'detectors': list(map(repr, detectors)),
'motor': repr(motor),
'start': start,
'stop': stop,
'min_step': min_step,
'max_step': max_step,
'target_delta': target_delta,
'backstep': backstep,
'threshold': threshold},
'plan_name': 'adaptive_scan',
'hints': {},
}
_md.update(md or {})
try:
dimensions = [(motor.hints['fields'], 'primary')]
except (AttributeError, KeyError):
pass
else:
_md['hints'].setdefault('dimensions', dimensions)
@bpp.stage_decorator(list(detectors) + [motor])
@bpp.run_decorator(md=_md)
def adaptive_core():
next_pos = start
step = (max_step - min_step) / 2
past_I = None
cur_I = None
cur_det = {}
if stop >= start:
direction_sign = 1
else:
direction_sign = -1
while next_pos * direction_sign < stop * direction_sign:
yield Msg('checkpoint')
yield from bps.mv(motor, next_pos)
yield Msg('create', None, name='primary')
for det in detectors:
yield Msg('trigger', det, group='B')
yield Msg('wait', None, 'B')
for det in utils.separate_devices(detectors + [motor]):
cur_det = yield Msg('read', det)
if target_field in cur_det:
cur_I = cur_det[target_field]['value']
yield Msg('save')
# special case first first loop
if past_I is None:
past_I = cur_I
next_pos += step * direction_sign
continue
dI = np.abs(cur_I - past_I)
slope = dI / step
if slope:
new_step = np.clip(target_delta / slope, min_step, max_step)
else:
new_step = np.min([step * 1.1, max_step])
# if we over stepped, go back and try again
if backstep and (new_step < step * threshold):
next_pos -= step
step = new_step
else:
past_I = cur_I
step = 0.2 * new_step + 0.8 * step
next_pos += step * direction_sign
return (yield from adaptive_core())
[docs]def rel_adaptive_scan(detectors, target_field, motor, start, stop,
min_step, max_step, target_delta, backstep,
threshold=0.8, *, md=None):
"""
Relative scan over one variable with adaptively tuned step size.
Parameters
----------
detectors : list
list of 'readable' objects
target_field : string
data field whose output is the focus of the adaptive tuning
motor : object
any 'settable' object (motor, temp controller, etc.)
start : float
starting position of motor
stop : float
ending position of motor
min_step : float
smallest step for fast-changing regions
max_step : float
largest step for slow-chaning regions
target_delta : float
desired fractional change in detector signal between steps
backstep : bool
whether backward steps are allowed -- this is concern with some motors
threshold : float, optional
threshold for going backward and rescanning a region, default is 0.8
md : dict, optional
metadata
See Also
--------
:func:`bluesky.plans.adaptive_scan`
"""
_md = {'plan_name': 'rel_adaptive_scan'}
_md.update(md or {})
@bpp.reset_positions_decorator([motor])
@bpp.relative_set_decorator([motor])
def inner_relative_adaptive_scan():
return (yield from adaptive_scan(detectors, target_field,
motor, start, stop, min_step,
max_step, target_delta,
backstep, threshold, md=_md))
return (yield from inner_relative_adaptive_scan())
[docs]def tune_centroid(
detectors, signal, motor,
start, stop, min_step,
num=10,
step_factor=3.0,
snake=False,
*, md=None):
r"""
plan: tune a motor to the centroid of signal(motor)
Initially, traverse the range from start to stop with
the number of points specified. Repeat with progressively
smaller step size until the minimum step size is reached.
Rescans will be centered on the signal centroid
(for $I(x)$, centroid$= \sum{I}/\sum{x*I}$)
with original scan range reduced by ``step_factor``.
Set ``snake=True`` if your positions are reproducible
moving from either direction. This will not necessarily
decrease the number of traversals required to reach convergence.
Snake motion reduces the total time spent on motion
to reset the positioner. For some positioners, such as
those with hysteresis, snake scanning may not be appropriate.
For such positioners, always approach the positions from the
same direction.
Note: Ideally the signal has only one peak in the range to
be scanned. It is assumed the signal is not polymodal
between ``start`` and ``stop``.
Parameters
----------
detectors : Signal
list of 'readable' objects
signal : string
detector field whose output is to maximize
motor : object
any 'settable' object (motor, temp controller, etc.)
start : float
start of range
stop : float
end of range, note: start < stop
min_step : float
smallest step size to use.
num : int, optional
number of points with each traversal, default = 10
step_factor : float, optional
used in calculating new range after each pass
note: step_factor > 1.0, default = 3
snake : bool, optional
if False (default), always scan from start to stop
md : dict, optional
metadata
Examples
--------
Find the center of a peak using synthetic hardware.
>>> from ophyd.sim import SynAxis, SynGauss
>>> motor = SynAxis(name='motor')
>>> det = SynGauss(name='det', motor, 'motor',
... center=-1.3, Imax=1e5, sigma=0.05)
>>> RE(tune_centroid([det], "det", motor, -1.5, -0.5, 0.01, 10))
"""
if min_step <= 0:
raise ValueError("min_step must be positive")
if step_factor <= 1.0:
raise ValueError("step_factor must be greater than 1.0")
try:
motor_name, = motor.hints['fields']
except (AttributeError, ValueError):
motor_name = motor.name
_md = {'detectors': [det.name for det in detectors],
'motors': [motor.name],
'plan_args': {'detectors': list(map(repr, detectors)),
'motor': repr(motor),
'start': start,
'stop': stop,
'num': num,
'min_step': min_step, },
'plan_name': 'tune_centroid',
'hints': {},
}
_md.update(md or {})
try:
dimensions = [(motor.hints['fields'], 'primary')]
except (AttributeError, KeyError):
pass
else:
_md['hints'].setdefault('dimensions', dimensions)
low_limit = min(start, stop)
high_limit = max(start, stop)
@bpp.stage_decorator(list(detectors) + [motor])
@bpp.run_decorator(md=_md)
def _tune_core(start, stop, num, signal):
next_pos = start
step = (stop - start) / (num - 1)
peak_position = None
cur_I = None
sum_I = 0 # for peak centroid calculation, I(x)
sum_xI = 0
while abs(step) >= min_step and low_limit <= next_pos <= high_limit:
yield Msg('checkpoint')
yield from bps.mv(motor, next_pos)
ret = (yield from bps.trigger_and_read(detectors + [motor]))
cur_I = ret[signal]['value']
sum_I += cur_I
position = ret[motor_name]['value']
sum_xI += position * cur_I
next_pos += step
in_range = min(start, stop) <= next_pos <= max(start, stop)
if not in_range:
if sum_I == 0:
return
peak_position = sum_xI / sum_I # centroid
sum_I, sum_xI = 0, 0 # reset for next pass
new_scan_range = (stop - start) / step_factor
start = np.clip(peak_position - new_scan_range/2,
low_limit, high_limit)
stop = np.clip(peak_position + new_scan_range/2,
low_limit, high_limit)
if snake:
start, stop = stop, start
step = (stop - start) / (num - 1)
next_pos = start
# print("peak position = {}".format(peak_position))
# print("start = {}".format(start))
# print("stop = {}".format(stop))
# finally, move to peak position
if peak_position is not None:
# improvement: report final peak_position
# print("final position = {}".format(peak_position))
yield from bps.mv(motor, peak_position)
return (yield from _tune_core(start, stop, num, signal))
[docs]def scan_nd(detectors, cycler, *, per_step=None, md=None):
"""
Scan over an arbitrary N-dimensional trajectory.
Parameters
----------
detectors : list
cycler : Cycler
cycler.Cycler object mapping movable interfaces to positions
per_step : callable, optional
hook for customizing action of inner loop (messages per step).
See docstring of :func:`bluesky.plan_stubs.one_nd_step` (the default)
for details.
md : dict, optional
metadata
See Also
--------
:func:`bluesky.plans.inner_product_scan`
:func:`bluesky.plans.grid_scan`
Examples
--------
>>> from cycler import cycler
>>> cy = cycler(motor1, [1, 2, 3]) * cycler(motor2, [4, 5, 6])
>>> scan_nd([sensor], cy)
"""
_md = {'detectors': [det.name for det in detectors],
'motors': [motor.name for motor in cycler.keys],
'num_points': len(cycler),
'num_intervals': len(cycler) - 1,
'plan_args': {'detectors': list(map(repr, detectors)),
'cycler': repr(cycler),
'per_step': repr(per_step)},
'plan_name': 'scan_nd',
'hints': {},
}
_md.update(md or {})
try:
dimensions = [(motor.hints['fields'], 'primary')
for motor in cycler.keys]
except (AttributeError, KeyError):
# Not all motors provide a 'fields' hint, so we have to skip it.
pass
else:
# We know that hints exists. Either:
# - the user passed it in and we are extending it
# - the user did not pass it in and we got the default {}
# If the user supplied hints includes a dimension entry, do not
# change it, else set it to the one generated above
_md['hints'].setdefault('dimensions', dimensions)
if per_step is None:
per_step = bps.one_nd_step
else:
# Ensure that the user-defined per-step has the expected signature.
sig = inspect.signature(per_step)
def _verify_1d_step(sig):
if len(sig.parameters) < 3:
return False
for name, (p_name, p) in zip_longest(['detectors', 'motor', 'step'], sig.parameters.items()):
# this is one of the first 3 positional arguements, check that the name matches
if name is not None:
if name != p_name:
return False
# if there are any extra arguments, check that they have a default
else:
if p.kind is p.VAR_KEYWORD or p.kind is p.VAR_POSITIONAL:
continue
if p.default is p.empty:
return False
return True
def _verify_nd_step(sig):
if len(sig.parameters) < 3:
return False
for name, (p_name, p) in zip_longest(['detectors', 'step', 'pos_cache'], sig.parameters.items()):
# this is one of the first 3 positional arguements, check that the name matches
if name is not None:
if name != p_name:
return False
# if there are any extra arguments, check that they have a default
else:
if p.kind is p.VAR_KEYWORD or p.kind is p.VAR_POSITIONAL:
continue
if p.default is p.empty:
return False
return True
if sig == inspect.signature(bps.one_nd_step):
pass
elif _verify_nd_step(sig):
# check other signature for back-compatibility
pass
elif _verify_1d_step(sig):
# Accept this signature for back-compat reasons (because
# inner_product_scan was renamed scan).
dims = len(list(cycler.keys))
if dims != 1:
raise TypeError("Signature of per_step assumes 1D trajectory "
"but {} motors are specified.".format(dims))
motor, = cycler.keys
user_per_step = per_step
def adapter(detectors, step, pos_cache):
# one_nd_step 'step' parameter is a dict; one_id_step 'step'
# parameter is a value
step, = step.values()
return (yield from user_per_step(detectors, motor, step))
per_step = adapter
else:
raise TypeError("per_step must be a callable with the signature \n "
"<Signature (detectors, step, pos_cache)> or "
"<Signature (detectors, motor, step)>. \n"
"per_step signature received: {}".format(sig))
pos_cache = defaultdict(lambda: None) # where last position is stashed
cycler = utils.merge_cycler(cycler)
motors = list(cycler.keys)
@bpp.stage_decorator(list(detectors) + motors)
@bpp.run_decorator(md=_md)
def inner_scan_nd():
for step in list(cycler):
yield from per_step(detectors, step, pos_cache)
return (yield from inner_scan_nd())
def inner_product_scan(detectors, num, *args, per_step=None, md=None):
# For scan, num is the _last_ positional arg instead of the first one.
# Notice the swapped order here.
md = md or {}
md.setdefault('plan_name', 'inner_product_scan')
yield from scan(detectors, *args, num, per_step=None, md=md)
[docs]def scan(detectors, *args, num=None, per_step=None, md=None):
"""
Scan over one multi-motor trajectory.
Parameters
----------
detectors : list
list of 'readable' objects
*args :
For one dimension, ``motor, start, stop``.
In general:
.. code-block:: python
motor1, start1, stop1,
motor2, start2, start2,
...,
motorN, startN, stopN
Motors can be any 'settable' object (motor, temp controller, etc.)
num : integer
number of points
per_step : callable, optional
hook for customizing action of inner loop (messages per step).
See docstring of :func:`bluesky.plan_stubs.one_nd_step` (the default)
for details.
md : dict, optional
metadata
See Also
--------
:func:`bluesky.plans.relative_inner_product_scan`
:func:`bluesky.plans.grid_scan`
:func:`bluesky.plans.scan_nd`
"""
# For back-compat reasons, we accept 'num' as the last positional argument:
# scan(detectors, motor, -1, 1, 3)
# or by keyword:
# scan(detectors, motor, -1, 1, num=3)
# ... which requires some special processing.
if num is None:
if len(args) % 3 != 1:
raise ValueError("The number of points to scan must be provided "
"as the last positional argument or as keyword "
"argument 'num'.")
num = args[-1]
args = args[:-1]
if not (float(num).is_integer() and num > 0.0):
raise ValueError(f"The parameter `num` is expected to be a number of "
f"steps (not step size!) It must therefore be a "
f"whole number. The given value was {num}.")
num = int(num)
md_args = list(chain(*((repr(motor), start, stop)
for motor, start, stop in partition(3, args))))
motor_names = tuple(motor.name for motor, start, stop
in partition(3, args))
md = md or {}
_md = {'plan_args': {'detectors': list(map(repr, detectors)),
'num': num, 'args': md_args,
'per_step': repr(per_step)},
'plan_name': 'scan',
'plan_pattern': 'inner_product',
'plan_pattern_module': plan_patterns.__name__,
'plan_pattern_args': dict(num=num, args=md_args),
'motors': motor_names
}
_md.update(md)
# get hints for best effort callback
motors = [motor for motor, start, stop in partition(3, args)]
# Give a hint that the motors all lie along the same axis
# [(['motor1', 'motor2', ...], 'primary'), ] is 1D (this case)
# [ ('motor1', 'primary'), ('motor2', 'primary'), ... ] is 2D for example
# call x_fields because these are meant to be the x (independent) axis
x_fields = []
for motor in motors:
x_fields.extend(getattr(motor, 'hints', {}).get('fields', []))
default_dimensions = [(x_fields, 'primary')]
default_hints = {}
if len(x_fields) > 0:
default_hints.update(dimensions=default_dimensions)
# now add default_hints and override any hints from the original md (if
# exists)
_md['hints'] = default_hints
_md['hints'].update(md.get('hints', {}) or {})
full_cycler = plan_patterns.inner_product(num=num, args=args)
return (yield from scan_nd(detectors, full_cycler,
per_step=per_step, md=_md))
[docs]def grid_scan(detectors, *args, snake_axes=None, per_step=None, md=None):
"""
Scan over a mesh; each motor is on an independent trajectory.
Parameters
----------
detectors: list
list of 'readable' objects
``*args``
patterned like (``motor1, start1, stop1, num1,``
``motor2, start2, stop2, num2,``
``motor3, start3, stop3, num3,`` ...
``motorN, startN, stopN, numN``)
The first motor is the "slowest", the outer loop. For all motors
except the first motor, there is a "snake" argument: a boolean
indicating whether to following snake-like, winding trajectory or a
simple left-to-right trajectory.
snake_axes: boolean or iterable, optional
which axes should be snaked, either ``False`` (do not snake any axes),
``True`` (snake all axes) or a list of axes to snake. "Snaking" an axis
is defined as following snake-like, winding trajectory instead of a
simple left-to-right trajectory. The elements of the list are motors
that are listed in `args`. The list must not contain the slowest
(first) motor, since it can't be snaked.
per_step: callable, optional
hook for customizing action of inner loop (messages per step).
See docstring of :func:`bluesky.plan_stubs.one_nd_step` (the default)
for details.
md: dict, optional
metadata
See Also
--------
:func:`bluesky.plans.rel_grid_scan`
:func:`bluesky.plans.inner_product_scan`
:func:`bluesky.plans.scan_nd`
"""
# Notes: (not to be included in the documentation)
# The deprecated function call with no 'snake_axes' argument and 'args'
# patterned like (``motor1, start1, stop1, num1,``
# ``motor2, start2, stop2, num2, snake2,``
# ``motor3, start3, stop3, num3, snake3,`` ...
# ``motorN, startN, stopN, numN, snakeN``)
# The first motor is the "slowest", the outer loop. For all motors
# except the first motor, there is a "snake" argument: a boolean
# indicating whether to following snake-like, winding trajectory or a
# simple left-to-right trajectory.
# Ideally, deprecated and new argument lists should not be mixed.
# The function will still accept `args` in the old format even if `snake_axes` is
# supplied, but if `snake_axes` is not `None` (the default value), it overrides
# any values of `snakeX` in `args`.
args_pattern = plan_patterns.classify_outer_product_args_pattern(args)
if (snake_axes is not None) and \
(args_pattern == plan_patterns.OuterProductArgsPattern.PATTERN_2):
raise ValueError("Mixing of deprecated and new API interface is not allowed: "
"the parameter 'snake_axes' can not be used if snaking is "
"set as part of 'args'")
# For consistency, set 'snake_axes' to False if new API call is detected
if (snake_axes is None) and \
(args_pattern != plan_patterns.OuterProductArgsPattern.PATTERN_2):
snake_axes = False
chunk_args = list(plan_patterns.chunk_outer_product_args(args, args_pattern))
# 'chunk_args' is a list of tuples of the form: (motor, start, stop, num, snake)
# If the function is called using deprecated pattern for arguments, then
# 'snake' may be set True for some motors, otherwise the 'snake' is always False.
# The list of controlled motors
motors = [_[0] for _ in chunk_args]
# Check that the same motor is not listed multiple times. This indicates an error in the script.
if len(set(motors)) != len(motors):
raise ValueError(f"Some motors are listed multiple times in the argument list 'args': "
f"'{motors}'")
if snake_axes is not None:
def _set_snaking(chunk, value):
"""Returns the tuple `chunk` with modified 'snake' value"""
_motor, _start, _stop, _num, _snake = chunk
return _motor, _start, _stop, _num, value
if isinstance(snake_axes, collections.abc.Iterable) and not isinstance(snake_axes, str):
# Always convert to a tuple (in case a `snake_axes` is an iterator).
snake_axes = tuple(snake_axes)
# Check if the list of axes (motors) contains repeated entries.
if len(set(snake_axes)) != len(snake_axes):
raise ValueError(f"The list of axes 'snake_axes' contains repeated elements: "
f"'{snake_axes}'")
# Check if the snaking is enabled for the slowest motor.
if len(motors) and (motors[0] in snake_axes):
raise ValueError(f"The list of axes 'snake_axes' contains the slowest motor: "
f"'{snake_axes}'")
# Check that all motors in the chunk_args are controlled in the scan.
# It is very likely that the script running the plan has a bug.
if any([_ not in motors for _ in snake_axes]):
raise ValueError(f"The list of axes 'snake_axes' contains motors "
f"that are not controlled during the scan: "
f"'{snake_axes}'")
# Enable snaking for the selected axes.
# If the argument `snake_axes` is specified (not None), then
# any `snakeX` values that could be specified in `args` are ignored.
for n, chunk in enumerate(chunk_args):
if n > 0: # The slowest motor is never snaked
motor = chunk[0]
if motor in snake_axes:
chunk_args[n] = _set_snaking(chunk, True)
else:
chunk_args[n] = _set_snaking(chunk, False)
elif snake_axes is True: # 'snake_axes' has boolean value `True`
# Set all 'snake' values except for the slowest motor
chunk_args = [_set_snaking(_, True) if n > 0 else _
for n, _ in enumerate(chunk_args)]
elif snake_axes is False: # 'snake_axes' has boolean value `True`
# Set all 'snake' values
chunk_args = [_set_snaking(_, False) for _ in chunk_args]
else:
raise ValueError(f"Parameter 'snake_axes' is not iterable, boolean or None: "
f"'{snake_axes}', type: {type(snake_axes)}")
# Prepare the argument list for the `outer_product` function
args_modified = []
for n, chunk in enumerate(chunk_args):
if n == 0:
args_modified.extend(chunk[:-1])
else:
args_modified.extend(chunk)
full_cycler = plan_patterns.outer_product(args=args_modified)
md_args = []
motor_names = []
motors = []
for i, (motor, start, stop, num, snake) in enumerate(chunk_args):
md_args.extend([repr(motor), start, stop, num])
if i > 0:
# snake argument only shows up after the first motor
md_args.append(snake)
motor_names.append(motor.name)
motors.append(motor)
_md = {'shape': tuple(num for motor, start, stop, num, snake
in chunk_args),
'extents': tuple([start, stop] for motor, start, stop, num, snake
in chunk_args),
'snaking': tuple(snake for motor, start, stop, num, snake
in chunk_args),
# 'num_points': inserted by scan_nd
'plan_args': {'detectors': list(map(repr, detectors)),
'args': md_args,
'per_step': repr(per_step)},
'plan_name': 'grid_scan',
'plan_pattern': 'outer_product',
'plan_pattern_args': dict(args=md_args),
'plan_pattern_module': plan_patterns.__name__,
'motors': tuple(motor_names),
'hints': {},
}
_md.update(md or {})
_md['hints'].setdefault('gridding', 'rectilinear')
try:
_md['hints'].setdefault('dimensions', [(m.hints['fields'], 'primary')
for m in motors])
except (AttributeError, KeyError):
...
return (yield from scan_nd(detectors, full_cycler,
per_step=per_step, md=_md))
[docs]def rel_grid_scan(detectors, *args, snake_axes=None, per_step=None, md=None):
"""
Scan over a mesh relative to current position.
Parameters
----------
detectors: list
list of 'readable' objects
``*args``
patterned like (``motor1, start1, stop1, num1,``
``motor2, start2, stop2, num2,``
``motor3, start3, stop3, num3,`` ...
``motorN, startN, stopN, numN``)
The first motor is the "slowest", the outer loop. For all motors
except the first motor, there is a "snake" argument: a boolean
indicating whether to following snake-like, winding trajectory or a
simple left-to-right trajectory.
snake_axes: boolean or iterable, optional
which axes should be snaked, either ``False`` (do not snake any axes),
``True`` (snake all axes) or a list of axes to snake. "Snaking" an axis
is defined as following snake-like, winding trajectory instead of a
simple left-to-right trajectory. The elements of the list are motors
that are listed in `args`. The list must not contain the slowest
(first) motor, since it can't be snaked.
per_step: callable, optional
hook for customizing action of inner loop (messages per step).
See docstring of :func:`bluesky.plan_stubs.one_nd_step` (the default)
for details.
md: dict, optional
metadata
See Also
--------
:func:`bluesky.plans.relative_inner_product_scan`
:func:`bluesky.plans.grid_scan`
:func:`bluesky.plans.scan_nd`
"""
# Notes: the deprecated function call is also supported. See the notes
# following the docstring for 'grid_scan' function
_md = {'plan_name': 'rel_grid_scan'}
_md.update(md or {})
motors = [m[0] for m in
plan_patterns.chunk_outer_product_args(args)]
@bpp.reset_positions_decorator(motors)
@bpp.relative_set_decorator(motors)
def inner_rel_grid_scan():
return (yield from grid_scan(detectors, *args,
snake_axes=snake_axes,
per_step=per_step, md=_md))
return (yield from inner_rel_grid_scan())
def relative_inner_product_scan(detectors, num, *args, per_step=None, md=None):
# For rel_scan, num is the _last_ positional arg instead of the first one.
# Notice the swapped order here.
md = md or {}
md.setdefault('plan_name', 'relative_inner_product_scan')
yield from rel_scan(detectors, *args, num, per_step=per_step, md=md)
[docs]def rel_scan(detectors, *args, num=None, per_step=None, md=None):
"""
Scan over one multi-motor trajectory relative to current position.
Parameters
----------
detectors : list
list of 'readable' objects
*args :
For one dimension, ``motor, start, stop``.
In general:
.. code-block:: python
motor1, start1, stop1,
motor2, start2, start2,
...,
motorN, startN, stopN,
Motors can be any 'settable' object (motor, temp controller, etc.)
num : integer
number of points
per_step : callable, optional
hook for customizing action of inner loop (messages per step).
See docstring of :func:`bluesky.plan_stubs.one_nd_step` (the default)
for details.
md : dict, optional
metadata
See Also
--------
:func:`bluesky.plans.rel_grid_scan`
:func:`bluesky.plans.inner_product_scan`
:func:`bluesky.plans.scan_nd`
"""
_md = {'plan_name': 'rel_scan'}
md = md or {}
_md.update(md)
motors = [motor for motor, start, stop in partition(3, args)]
@bpp.reset_positions_decorator(motors)
@bpp.relative_set_decorator(motors)
def inner_rel_scan():
return (yield from scan(detectors, *args, num=num,
per_step=per_step, md=_md))
return (yield from inner_rel_scan())
[docs]def tweak(detector, target_field, motor, step, *, md=None):
"""
Move and motor and read a detector with an interactive prompt.
Parameters
----------
detetector : Device
target_field : string
data field whose output is the focus of the adaptive tuning
motor : Device
step : float
initial suggestion for step size
md : dict, optional
metadata
"""
prompt_str = '{0}, {1:.3}, {2:.3}, ({3}) '
_md = {'detectors': [detector.name],
'motors': [motor.name],
'plan_args': {'detector': repr(detector),
'target_field': target_field,
'motor': repr(motor),
'step': step},
'plan_name': 'tweak',
'hints': {},
}
try:
dimensions = [(motor.hints['fields'], 'primary')]
except (AttributeError, KeyError):
pass
else:
_md['hints'].update({'dimensions': dimensions})
_md.update(md or {})
d = detector
try:
from IPython.display import clear_output
except ImportError:
# Define a no-op for clear_output.
def clear_output(wait=False):
pass
@bpp.stage_decorator([detector, motor])
@bpp.run_decorator(md=_md)
def tweak_core():
nonlocal step
while True:
yield Msg('create', None, name='primary')
ret_mot = yield Msg('read', motor)
if ret_mot is None:
return
key = list(ret_mot.keys())[0]
pos = ret_mot[key]['value']
yield Msg('trigger', d, group='A')
yield Msg('wait', None, 'A')
reading = yield Msg('read', d)
val = reading[target_field]['value']
yield Msg('save')
prompt = prompt_str.format(motor.name, float(pos),
float(val), step)
new_step = yield Msg('input', prompt=prompt)
if new_step:
try:
step = float(new_step)
except ValueError:
break
yield Msg('set', motor, pos + step, group='A')
print('Motor moving...')
sys.stdout.flush()
yield Msg('wait', None, 'A')
clear_output(wait=True)
# stackoverflow.com/a/12586667/380231
print('\x1b[1A\x1b[2K\x1b[1A')
return (yield from tweak_core())
[docs]def spiral_fermat(detectors, x_motor, y_motor, x_start, y_start, x_range,
y_range, dr, factor, *, dr_y=None, tilt=0.0, per_step=None,
md=None):
'''Absolute fermat spiral scan, centered around (x_start, y_start)
Parameters
----------
detectors : list
list of 'readable' objects
x_motor : object
any 'settable' object (motor, temp controller, etc.)
y_motor : object
any 'settable' object (motor, temp controller, etc.)
x_start : float
x center
y_start : float
y center
x_range : float
x width of spiral
y_range : float
y width of spiral
dr : float
delta radius
factor : float
radius gets divided by this
dr_y : float, optional
Delta radius along the major axis of the ellipse, if not specifed
defaults to dr.
tilt : float, optional
Tilt angle in radians, default 0.0
per_step : callable, optional
hook for customizing action of inner loop (messages per step).
See docstring of :func:`bluesky.plan_stubs.one_nd_step` (the default)
for details.
md : dict, optional
metadata
See Also
--------
:func:`bluesky.plans.spiral`
:func:`bluesky.plans.rel_spiral`
:func:`bluesky.plans.rel_spiral_fermat`
'''
pattern_args = dict(x_motor=x_motor, y_motor=y_motor, x_start=x_start,
y_start=y_start, x_range=x_range, y_range=y_range,
dr=dr, factor=factor, dr_y=dr_y, tilt=tilt)
cyc = plan_patterns.spiral_fermat(**pattern_args)
# Before including pattern_args in metadata, replace objects with reprs.
pattern_args['x_motor'] = repr(x_motor)
pattern_args['y_motor'] = repr(y_motor)
_md = {'plan_args': {'detectors': list(map(repr, detectors)),
'x_motor': repr(x_motor), 'y_motor': repr(y_motor),
'x_start': x_start, 'y_start': y_start,
'x_range': x_range, 'y_range': y_range,
'dr': dr, 'factor': factor, 'dr_y': dr_y,
'tilt': tilt, 'per_step': repr(per_step)},
'extents': tuple([[x_start - x_range, x_start + x_range],
[y_start - y_range, y_start + y_range]]),
'plan_name': 'spiral_fermat',
'plan_pattern': 'spiral_fermat',
'plan_pattern_module': plan_patterns.__name__,
'plan_pattern_args': pattern_args,
'hints': {},
}
try:
dimensions = [(x_motor.hints['fields'], 'primary'),
(y_motor.hints['fields'], 'primary')]
except (AttributeError, KeyError):
pass
else:
_md['hints'].update({'dimensions': dimensions})
_md.update(md or {})
return (yield from scan_nd(detectors, cyc, per_step=per_step, md=_md))
[docs]def rel_spiral_fermat(detectors, x_motor, y_motor, x_range, y_range, dr,
factor, *, dr_y=None, tilt=0.0, per_step=None, md=None):
'''Relative fermat spiral scan
Parameters
----------
detectors : list
list of 'readable' objects
x_motor : object
any 'settable' object (motor, temp controller, etc.)
y_motor : object
any 'settable' object (motor, temp controller, etc.)
x_range : float
x width of spiral
y_range : float
y width of spiral
dr : float
delta radius
factor : float
radius gets divided by this
dr_y : float, optional
Delta radius along the major axis of the ellipse, if not specifed
defaults to dr
tilt : float, optional
Tilt angle in radians, default 0.0
per_step : callable, optional
hook for customizing action of inner loop (messages per step).
See docstring of :func:`bluesky.plan_stubs.one_nd_step` (the default)
for details.
md : dict, optional
metadata
See Also
--------
:func:`bluesky.plans.spiral`
:func:`bluesky.plans.rel_spiral`
:func:`bluesky.plans.spiral_fermat`
'''
_md = {'plan_name': 'rel_spiral_fermat'}
_md.update(md or {})
@bpp.reset_positions_decorator([x_motor, y_motor])
@bpp.relative_set_decorator([x_motor, y_motor])
def inner_relative_spiral_fermat():
return (yield from spiral_fermat(detectors, x_motor, y_motor,
0, 0,
x_range, y_range,
dr, factor, dr_y=dr_y, tilt=tilt,
per_step=per_step, md=_md))
return (yield from inner_relative_spiral_fermat())
[docs]def spiral(detectors, x_motor, y_motor, x_start, y_start, x_range, y_range, dr,
nth, *, dr_y=None, tilt=0.0, per_step=None, md=None):
'''Spiral scan, centered around (x_start, y_start)
Parameters
----------
x_motor : object
any 'settable' object (motor, temp controller, etc.)
y_motor : object
any 'settable' object (motor, temp controller, etc.)
x_start : float
x center
y_start : float
y center
x_range : float
x width of spiral
y_range : float
y width of spiral
dr : float
Delta radius along the minor axis of the ellipse.
dr_y : float, optional
Delta radius along the major axis of the ellipse. If None, defaults to
dr.
nth : float
Number of theta steps
tilt : float, optional
Tilt angle in radians, default 0.0
per_step : callable, optional
hook for customizing action of inner loop (messages per step).
See docstring of :func:`bluesky.plan_stubs.one_nd_step` (the default)
for details.
md : dict, optional
metadata
See Also
--------
:func:`bluesky.plans.rel_spiral`
:func:`bluesky.plans.spiral_fermat`
:func:`bluesky.plans.rel_spiral_fermat`
'''
pattern_args = dict(x_motor=x_motor, y_motor=y_motor, x_start=x_start,
y_start=y_start, x_range=x_range, y_range=y_range,
dr=dr, nth=nth, dr_y=dr_y, tilt=tilt)
cyc = plan_patterns.spiral(**pattern_args)
# Before including pattern_args in metadata, replace objects with reprs.
pattern_args['x_motor'] = repr(x_motor)
pattern_args['y_motor'] = repr(y_motor)
_md = {'plan_args': {'detectors': list(map(repr, detectors)),
'x_motor': repr(x_motor), 'y_motor': repr(y_motor),
'x_start': x_start, 'y_start': y_start,
'x_range': x_range, 'y_range': y_range,
'dr': dr, 'dr_y': dr_y, 'nth': nth, 'tilt': tilt,
'per_step': repr(per_step)},
'extents': tuple([[x_start - x_range, x_start + x_range],
[y_start - y_range, y_start + y_range]]),
'plan_name': 'spiral',
'plan_pattern': 'spiral',
'plan_pattern_args': pattern_args,
'plan_pattern_module': plan_patterns.__name__,
'hints': {},
}
try:
dimensions = [(x_motor.hints['fields'], 'primary'),
(y_motor.hints['fields'], 'primary')]
except (AttributeError, KeyError):
pass
else:
_md['hints'].update({'dimensions': dimensions})
_md.update(md or {})
return (yield from scan_nd(detectors, cyc, per_step=per_step, md=_md))
[docs]def rel_spiral(detectors, x_motor, y_motor, x_range, y_range, dr, nth,
*, dr_y=None, tilt=0.0, per_step=None, md=None):
'''Relative spiral scan
Parameters
----------
x_motor : object
any 'settable' object (motor, temp controller, etc.)
y_motor : object
any 'settable' object (motor, temp controller, etc.)
x_range : float
x width of spiral
y_range : float
y width of spiral
dr : float
Delta radius along the minor axis of the ellipse.
dr_y : float, optional
Delta radius along the major axis of the ellipse. If None, it
defaults to dr.
nth : float
Number of theta steps
tilt : float, optional
Tilt angle in radians, default 0.0
per_step : callable, optional
hook for customizing action of inner loop (messages per step).
See docstring of :func:`bluesky.plan_stubs.one_nd_step` (the default)
for details.
md : dict, optional
metadata
See Also
--------
:func:`bluesky.plans.spiral`
:func:`bluesky.plans.spiral_fermat`
'''
_md = {'plan_name': 'rel_spiral'}
_md.update(md or {})
@bpp.reset_positions_decorator([x_motor, y_motor])
@bpp.relative_set_decorator([x_motor, y_motor])
def inner_relative_spiral():
return (yield from spiral(detectors, x_motor, y_motor,
0, 0,
x_range, y_range, dr, nth,
dr_y=dr_y, tilt=tilt,
per_step=per_step, md=_md))
return (yield from inner_relative_spiral())
[docs]def spiral_square(detectors, x_motor, y_motor, x_center, y_center, x_range,
y_range, x_num, y_num, *, per_step=None, md=None):
'''Absolute square spiral scan, centered around (x_center, y_center)
Parameters
----------
detectors : list
list of 'readable' objects
x_motor : object
any 'settable' object (motor, temp controller, etc.)
y_motor : object
any 'settable' object (motor, temp controller, etc.)
x_center : float
x center
y_center : float
y center
x_range : float
x width of spiral
y_range : float
y width of spiral
x_num : float
number of x axis points
y_num : float
Number of y axis points.
per_step : callable, optional
hook for cutomizing action of inner loop (messages per step).
See docstring of :func:`bluesky.plans.one_nd_step` (the default) for
details.
md : dict, optional
metadata
See Also
--------
:func:`bluesky.plans.relative_spiral_square`
:func:`bluesky.plans.spiral`
:func:`bluesky.plans.relative_spiral`
:func:`bluesky.plans.spiral_fermat`
:func:`bluesky.plans.relative_spiral_fermat`
'''
pattern_args = dict(x_motor=x_motor, y_motor=y_motor, x_center=x_center,
y_center=y_center, x_range=x_range, y_range=y_range,
x_num=x_num, y_num=y_num)
cyc = plan_patterns.spiral_square_pattern(**pattern_args)
# Before including pattern_args in metadata, replace objects with reprs.
pattern_args['x_motor'] = repr(x_motor)
pattern_args['y_motor'] = repr(y_motor)
_md = {'plan_args': {'detectors': list(map(repr, detectors)),
'x_motor': repr(x_motor), 'y_motor': repr(y_motor),
'x_center': x_center, 'y_center': y_center,
'x_range': x_range, 'y_range': y_range,
'x_num': x_num, 'y_num': y_num,
'per_step': repr(per_step)},
'plan_name': 'spiral_square',
'plan_pattern': 'spiral_square',
'shape': (y_num, x_num),
'extents': ((y_center - y_range / 2, y_center + y_range / 2),
(x_center - x_range / 2, x_center + x_range / 2)),
'hints': {},
}
_md.update(md or {})
_md['hints'].setdefault('gridding', 'rectilinear_nonsequential')
try:
_md['hints'].setdefault('dimensions', [(m.hints['fields'], 'primary')
for m in [y_motor, x_motor]])
except (AttributeError, KeyError):
...
return (yield from scan_nd(detectors, cyc, per_step=per_step, md=_md))
[docs]def rel_spiral_square(detectors, x_motor, y_motor, x_range, y_range,
x_num, y_num, *, per_step=None, md=None):
'''Relative square spiral scan, centered around current (x, y) position.
Parameters
----------
detectors : list
list of 'readable' objects
x_motor : object
any 'settable' object (motor, temp controller, etc.)
y_motor : object
any 'settable' object (motor, temp controller, etc.)
x_range : float
x width of spiral
y_range : float
y width of spiral
x_num : float
number of x axis points
y_num : float
Number of y axis points.
per_step : callable, optional
hook for cutomizing action of inner loop (messages per step).
See docstring of :func:`bluesky.plans.one_nd_step` (the default) for
details.
md : dict, optional
metadata
See Also
--------
:func:`bluesky.plans.spiral_square`
:func:`bluesky.plans.spiral`
:func:`bluesky.plans.relative_spiral`
:func:`bluesky.plans.spiral_fermat`
:func:`bluesky.plans.relative_spiral_fermat`
'''
_md = {'plan_name': 'rel_spiral_square'}
_md.update(md or {})
@bpp.reset_positions_decorator([x_motor, y_motor])
@bpp.relative_set_decorator([x_motor, y_motor])
def inner_relative_spiral():
return (yield from spiral_square(detectors, x_motor, y_motor,
0, 0,
x_range, y_range, x_num, y_num,
per_step=per_step, md=_md))
return (yield from inner_relative_spiral())
[docs]def ramp_plan(go_plan,
monitor_sig,
inner_plan_func,
take_pre_data=True,
timeout=None,
period=None, md=None):
'''Take data while ramping one or more positioners.
The pseudo code for this plan is ::
sts = (yield from go_plan)
yield from open_run()
yield from inner_plan_func()
while not st.done:
yield from inner_plan_func()
yield from inner_plan_func()
yield from close_run()
Parameters
----------
go_plan : generator
plan to start the ramp. This will be run inside of a open/close
run.
This plan must return a `ophyd.StatusBase` object.
inner_plan_func : generator function
generator which takes no input
This will be called for every data point. This should create
one or more events.
timeout : float, optional
If not None, the maximum time the ramp can run.
In seconds
take_pre_data: Bool, optional
If True, add a pre data at beginning
period : float, optional
If not None, take data no faster than this. If None, take
data as fast as possible
If running the inner plan takes longer than `period` than take
data with no dead time.
In seconds.
'''
_md = {'plan_name': 'ramp_plan'}
_md.update(md or {})
@bpp.monitor_during_decorator((monitor_sig,))
@bpp.run_decorator(md=_md)
def polling_plan():
fail_time = None
if timeout is not None:
# sort out if we should watch the clock
fail_time = time.time() + timeout
# take a 'pre' data point
if take_pre_data:
yield from inner_plan_func()
# start the ramp
status = (yield from go_plan)
while not status.done:
start_time = time.time()
yield from inner_plan_func()
if fail_time is not None:
if time.time() > fail_time:
raise utils.RampFail()
if period is not None:
cur_time = time.time()
wait_time = (start_time + period) - cur_time
if wait_time > 0:
yield from bps.sleep(wait_time)
# take a 'post' data point
yield from inner_plan_func()
return (yield from polling_plan())
[docs]def fly(flyers, *, md=None):
"""
Perform a fly scan with one or more 'flyers'.
Parameters
----------
flyers : collection
objects that support the flyer interface
md : dict, optional
metadata
Yields
------
msg : Msg
'kickoff', 'wait', 'complete, 'wait', 'collect' messages
See Also
--------
:func:`bluesky.preprocessors.fly_during_wrapper`
:func:`bluesky.preprocessors.fly_during_decorator`
"""
uid = yield from bps.open_run(md)
for flyer in flyers:
yield from bps.kickoff(flyer, wait=True)
for flyer in flyers:
yield from bps.complete(flyer, wait=True)
for flyer in flyers:
yield from bps.collect(flyer)
yield from bps.close_run()
return uid
def x2x_scan(detectors, motor1, motor2, start, stop, num, *,
per_step=None, md=None):
"""
Relatively scan over two motors in a 2:1 ratio
This is a generalized version of a theta2theta scan
Parameters
----------
detectors : list
list of 'readable' objects
motor1, motor2 : Positioner
The second motor will move half as much as the first
start, stop : float
The relative limits of the first motor. The second motor
will move between ``start / 2`` and ``stop / 2``
per_step : callable, optional
hook for cutomizing action of inner loop (messages per step).
See docstring of :func:`bluesky.plan_stubs.one_nd_step` (the default)
for details.
md : dict, optional
metadata
"""
_md = {'plan_name': 'x2x_scan',
'plan_args': {'detectors': list(map(repr, detectors)),
'motor1': motor1.name,
'motor2': motor2.name,
'start': start, 'stop': stop, 'num': num,
'per_step': repr(per_step)}
}
_md.update(md or {})
return (yield from relative_inner_product_scan(
detectors, num,
motor1, start, stop,
motor2, start / 2, stop / 2,
per_step=per_step,
md=_md))
relative_list_scan = rel_list_scan # back-compat
relative_scan = rel_scan # back-compat
relative_log_scan = rel_log_scan # back-compat
relative_adaptive_scan = rel_adaptive_scan # back-compat
outer_product_scan = grid_scan # back-compat
relative_outer_product_scan = rel_grid_scan # back-compat
relative_spiral_fermat = rel_spiral_fermat # back-compat
relative_spiral = rel_spiral # back-compat