import os
from collections import Iterable
from pathlib import Path
from bluesky.callbacks import CallbackBase
from xpdan.dev_utils import _timestampstr
[docs]def if_dark(doc):
return doc.get('dark_frame', False)
[docs]def if_query_results(n_hdrs):
return n_hdrs > 0
[docs]def if_calibration(start):
# print("detector cal tf ================================")
# pprint(start)
# print('detector_calibration_server_uid' in start)
# return 'is_calibration' in start
return 'detector_calibration_server_uid' in start
[docs]def if_not_calibration(doc):
return 'is_calibration' not in doc and 'calibration_md' in doc
# return 'calibration_client_uid' in doc
# return 'calibration_server_uid' not in doc
[docs]def dark_template_func(timestamp, template):
"""Format template for dark images
Parameters
----------
timestamp: float
The time in unix epoch
template: str
The string to be formatted
Returns
-------
str:
"""
d = {'human_timestamp': _timestampstr(timestamp), 'ext': '.tiff'}
t = template.format(**d)
os.makedirs(os.path.split(t)[0])
return t
[docs]def templater1_func(doc, template):
"""format base string with data from experiment, sample_name,
folder_tag"""
d = {'sample_name': doc.get('sample_name', ''),
'folder_tag': doc.get('folder_tag', '')}
return template.format(**d)
[docs]def templater2_func(doc, template, aux=None, short_aux=None):
"""format with auxiliary and time"""
if aux is None:
aux = ['temperature', 'diff_x', 'diff_y', 'eurotherm']
if short_aux is None:
short_aux = ['temp', 'x', 'y', 'euro']
aux_res = ['{}={}'.format(b, doc['data'].get(a, ''))
for a, b in zip(aux, short_aux)]
aux_res_str = '_'.join(aux_res)
# Add a separator between timestamp and extras
if aux_res_str:
aux_res_str = '_' + aux_res_str
return template.format(
# Change to include name as well
auxiliary=aux_res_str,
human_timestamp=_timestampstr(doc['time']))
[docs]def templater3_func(template, analysis_stage='raw', ext='.tiff'):
return Path(template.format(analysis_stage=analysis_stage,
ext=ext)).as_posix()
[docs]def clear_combine_latest(node, position=None):
if position is None:
position = range(len(node.last))
elif not isinstance(position, Iterable):
position = (position,)
for p in position:
node.last[p] = None
node.missing.add(node.upstreams[p])
[docs]class Filler(CallbackBase):
"""Fill events without provenence"""
def __init__(self, db):
self.db = db
self.descs = None
[docs] def start(self, docs):
self.descs = []
return 'start', docs
[docs] def descriptor(self, docs):
self.descs.append(docs)
return 'descriptor', docs
[docs] def event(self, docs):
d = next(self.db.fill_events([docs], self.descs))
return 'event', d
[docs] def stop(self, docs):
return 'stop', docs
base_template = (''
'{base_folder}/{folder_prefix}/'
'{analysis_stage}/'
'{raw_start[sample_name]}_'
'{human_timestamp}_'
'[temp_{raw_event[data][temperature]:1.2f}'
'{raw_descriptor[data_keys][temperature][units]}]_'
'[dx_{raw_event[data][diff_x]:1.3f}'
'{raw_descriptor[data_keys][diff_x][units]}]_'
'[dy_{raw_event[data][diff_y]:1.3f}'
'{raw_descriptor[data_keys][diff_y][units]}]_'
'{raw_start[uid]:.6}_'
'{raw_event[seq_num]:04d}{ext}')