Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- ## 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)
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement