Source code for xpdan.fuzzybroker

"""Enhanced databroker with fuzzy search options"""
##############################################################################
#
# xpdan            by Billinge Group
#                   Simon J. L. Billinge sb2896@columbia.edu
#                   (c) 2016 trustees of Columbia University in the City of
#                        New York.
#                   All rights reserved
#
# File coded by:    Christopher J. Wright
#
# See AUTHORS.txt for a list of people who contributed.
# See LICENSE.txt for license information.
#
##############################################################################

from heapq import heapify, heappushpop
from pprint import pprint
from pyxdameraulevenshtein import \
    normalized_damerau_levenshtein_distance as ndld

from databroker.broker import Broker
from .dev_utils import _timestampstr


[docs]class FuzzyBroker(Broker):
[docs] def beamtime_dates(self, **kwargs): """Get info for each beamtime Parameters ---------- keys: iterable of str The keys to be included in the return beamtime_key: str The key for the unique beamtime key print_results: bool If true prints the information Returns ------- list of dicts: The list of beamtimes and their associated information """ return beamtime_dates(self, **kwargs)
def _get_from_dict(data_dict, map_list): """Get a value from a nested dictionary, given a list of keys Parameters ---------- data_dict: dict The dictionary to be queried map_list: list of str A list of strings, each string is one level lower than the previous Returns ------- object: The the value from the dict """ for k in map_list: data_dict = data_dict[k] return data_dict def _nested_dict_values(d): """Yield all string values inside a nested dictionary Parameters ---------- d: dict The dictionary to be unpacked Yields ------- str: The string value inside the dictionary """ for v in d.values(): if isinstance(v, dict): yield from _nested_dict_values(v) else: if isinstance(v, str): yield v else: yield None
[docs]def beamtime_dates(db, keys=('beamtime_uid', 'bt_safN', 'facility', 'beamline'), beamtime_key='beamtime_uid', print_results=True): """Get info for each beamtime Parameters ---------- db: databroker instance The databroker to be searched keys: iterable of str The keys to be included in the return beamtime_key: str The key for the unique beamtime key print_results: bool If true prints the information Returns ------- list of dicts: The list of beamtimes and their associated information """ hdrs = db() bts = set([h['start'][beamtime_key] for h in hdrs]) returns = [] for s in bts: hdrs = db(**{beamtime_key: s}) start_hdr = next(iter(hdrs)) # XXX: this is bad but I have no other way to get the latest element # from results for hdr in hdrs: stop_hdr = hdr for i, hdr in enumerate(hdrs): if i == 0: start_hdr = hdr pass stop_hdr = hdr info = {k: start_hdr[k] for k in keys if k in start_hdr.keys()} info.update({'start_time': _timestampstr(start_hdr['start']['time']), 'stop_time': _timestampstr(stop_hdr['start']['time'])}) returns.append(info) if print_results: pprint(returns) return returns