Skip to content

Instantly share code, notes, and snippets.

Show Gist options
  • Save danielballan/e18cfded60e390908f47b12ea506e520 to your computer and use it in GitHub Desktop.
Save danielballan/e18cfded60e390908f47b12ea506e520 to your computer and use it in GitHub Desktop.
## customized count
def delayed_count(detectors, shutter_config, interval=None, num=1, *, md=None):
"""
Take one or more readings from detectors.
Parameters
----------
detectors : list
list of 'readable' objects
gasses : list of string
name of gasses want to alterna
shutter_config : tuple
a tuple of shutter configuration, expected to be (shutter, open_satus, close_status)
interval : float
FIXME
num : integer, optional
number of readings to take; default is 1
If None, capture data until canceled
md : dict, optional
metadata
Note
----
If ``delay`` is an iterable, it must have at least ``num - 1`` entries or
the plan will raise a ``ValueError`` during iteration.
"""
if md is None:
md = {}
md = ChainMap(
md,
{'detectors': [det.name for det in detectors],
'num_steps': num,
'plan_args': {'detectors': list(map(repr, detectors)), 'num': num},
'plan_name': 'count'})
# If delay is a scalar, repeat it forever. If it is an iterable, leave it.
#if not isinstance(delay, Iterable):
# delay = itertools.repeat(delay)
#else:
# delay = iter(delay)
# unpack shutter
shutter, sh_open, sh_close = shutter_config
@stage_decorator(detectors)
@run_decorator(md=md)
def finite_plan():
for i in range(num):
ts1 = time.time()
# light frame
yield Msg('checkpoint')
yield from trigger_and_read(detectors)
# dark frame
yield Msg('checkpoint')
yield from abs_set(shutter, sh_close)
yield from trigger_and_read(detectors)
yield from abs_set(shutter, sh_open)
# delay time
ts2 = time.time()
#try:
# d = next(delay)
#except StopIteration:
# if i + 1 == num:
# break
# else:
# # num specifies a number of iterations less than delay
# raise ValueError("num=%r but delays only provides %r "
# "entries" % (num, i))
if interval is not None:
yield Msg('sleep', None, ts1 + interval - ts2)
@stage_decorator(detectors)
@run_decorator(md=md)
def infinite_plan():
while True:
ts1 = time.time()
# light frame
yield Msg('checkpoint')
yield from trigger_and_read(detectors)
# dark frame
yield Msg('checkpoint')
yield from abs_set(shutter, sh_close)
yield from trigger_and_read(detectors)
yield from abs_set(shutter, sh_open)
#try:
# d = next(delay)
#except StopIteration:
# break
if interval is not None:
yield Msg('sleep', None, ts1 + interval - ts2)
if num is None:
return (yield from infinite_plan())
else:
return (yield from finite_plan())
## customized TiffExporter
class XpdLiveTiffExporter(LiveTiffExporter):
"""
subclass to customize functionality
Save TIFF files.
Incorporate metadata and data from individual data points in the filenames.
Parameters
----------
field : str
a data key, e.g., 'image'
template : str
A templated file path, where curly brackets will be filled in with
the attributes of 'start', 'event', and (for image stacks) 'i',
a sequential number.
e.g., "dir/scan{start.scan_id}_by_{start.experimenter}_{i}.tiff"
dryrun : bool
default to False; if True, do not write any files
overwrite : bool
default to False, raising an OSError if file exists
db : Broker, optional
The databroker instance to use, if not provided use databroker
singleton
Attributes
----------
filenames : list of filenames written in ongoing or most recent run
"""
def _save_image(self, image, filename):
fn_head, fn_tail = os.split(filename)
if not os.path.isdir(fn_head):
os.makedirs(fn_head, exist_ok=True)
super()._save_image(image, filename)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment