Here is a script demonstrating several features.
#!/usr/bin/env python3 """ This script demonstrates several bluesky features, see comment lines To be launched like this: ipython3 --matplotlib=qt5 scanManyFeatures.py """ from bluesky import Msg from bluesky.plans import scan import blueskyDESY from bluesky import RunEngine from bluesky.callbacks.best_effort import BestEffortCallback from bluesky.preprocessors import SupplementalData from databroker import Broker from pprint import pprint def dumpHeader( hdr): print( " --- hdr") print( " - uid %s" % repr( hdr.uid)) print( " - table() %s" % repr( hdr.table())) print( " - fields() %s" % repr( hdr.fields())) print( " - stream names %s" % repr( hdr.stream_names)) print( " - devices %s" % repr( hdr.devices())) for d in hdr.devices(): print( " dumpHeader: config data %s %s" % (d, repr( hdr.config_data( d)))) print( " dumpHeader: start, BEGIN") pprint( hdr.start) print( " dumpHeader: start, END") for des in hdr.descriptors: print( "dumpHeader: descriptor") pprint( des) for e in hdr.events(): print( "dumpHeader: event") pprint( e) print( " dumpHeader: stop, BEGIN") pprint( hdr.stop) print( " dumpHeader: stop, END") print( "") return def main(): db = Broker.named('temp') # # if Experiment() is called without read_attrs, # the activeMntGrp is used for device selection # mg = blueskyDESY.Experiment() mg_baseline = blueskyDESY.Experiment( read_attrs = [ 'eh_c10', 'eh_c11']) eh_mot70 = blueskyDESY.motorTango( name = 'eh_mot70') eh_mot71 = blueskyDESY.motorTango( name = 'eh_mot71') # # create a run engine # RE = RunEngine() # # take snapshots (baseline readings) of some hardware # # sd = SupplementalData() #RE.preprocessors.append(sd) sd.baseline = [mg_baseline, eh_mot71] # # insert all metadata/data into db # RE.subscribe( db.insert) # # matplotlib and printout # problems with MCAs # #bec = BestEffortCallback() #RE.subscribe( bec) # # data to pyspMonitor3.py # RE.subscribe(blueskyDESY.docCallback()) # # metadata, per RE and per plan # RE.md[ 'user'] = 'p09user' # permanent metadata, lifetime of RE uid, = RE(scan([mg], eh_mot70, 0, 0.1, 20, md={ 'sample':'Gold'})) # metadata for this run mg = blueskyDESY.Experiment( read_attrs = [ 'eh_c01', 'eh_c02', 'eh_c03', 'eh_c04', 'eh_c05', 'eh_mca01']) uid, = RE(scan([mg], eh_mot71, 0.1, 0.2, 20, md={ 'sample':'Silver'})) # metadata for this run #header = db[uid] header = db[-1] print( "\n*** dump header for db[ -1]") dumpHeader( header) # # locate data labeled 'p09user' # print( "\n*** dump header for p09user") res = db( user='p09user') for hdr in res: dumpHeader( hdr) # # locate data labled 'silver' # print( "\n*** dump header for silver") res = db( sample='Silver') for hdr in res: dumpHeader( hdr) return if __name__ == "__main__": main()