Source code for

__all__ = ["from_databroker"]

from ._data import Data

[docs]def from_databroker(run, dataset="primary"): """Import a dataset from a bluesky run into the WrightTools Data format. Parameters ---------- run: BlueskyRun The bluesky run as returned by e.g. catalog["<uid>"] dataset: str The string identifier of the stream to import from the bluesky run. By default "primary" is used, but e.g. "baseline" is also common """ describe = run.describe() md = describe["metadata"] start = md["start"] ds = run[dataset].read() shape = start.get("shape", (len(ds.time),)) detectors = start.get("detectors", []) data = Data(name=start["uid"]) for var in ds: if var == "uid": continue if var.endswith("_busy"): continue if any(var.startswith(d) for d in detectors): data.create_channel(var, values=ds[var].data.reshape(shape)) else: # TODO units, once they are in the dataset metadata data.create_variable(var, values=ds[var].data.reshape(shape)) transform = [x[0] for x, ds_name in start["hints"]["dimensions"] if ds_name == dataset] data.transform(*transform) return data