Skip to content

Instantly share code, notes, and snippets.

@danielballan

danielballan/sample_example.py

Last active September 5, 2018 21:32
Download ZIP

Revisions

  1. danielballan revised this gist Aug 21, 2018. 1 changed file with 3 additions and 2 deletions.
    5 changes: 3 additions & 2 deletions sample_example.py
    View file
    Original file line number Diff line number Diff line change
    @@ -1,5 +1,6 @@
    # Do this part whether you acquiring data or accessing it.
    import pymongo
    import pymongo
    cli = pymongo.MongoClient('xf11id-ca')
    samples = cli.get_database('samples').get_collection('samples')
    from databroker import Broker
    db = Broker.named('temp') # for real applications, 'temp' would be 'chx'
    @@ -34,4 +35,4 @@ def fetch_sample(start, stop):
    # This tells the databroker to addd some custom stuff to the `.ext` attribute
    # of each Header.
    db.external_fetchers['sample_info'] = fetch_sample
    print(db[-1].ext)
    print(db[-1].ext.sample_info)
  2. danielballan created this gist Aug 21, 2018.
    37 changes: 37 additions & 0 deletions sample_example.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,37 @@
    # Do this part whether you acquiring data or accessing it.
    import pymongo
    samples = cli.get_database('samples').get_collection('samples')
    from databroker import Broker
    db = Broker.named('temp') # for real applications, 'temp' would be 'chx'


    # Do this part when you are entering new samples.
    # I'm going to assume that 'name' is unique. We should be more careful about
    # ensuring that and maybe come up with a richer system, but this is good enough
    # for a toy exmaple.
    samples.insert(dict(name='A', composition='Ni'))
    samples.insert(dict(name='B', composition='Au'))


    # Here I import something things and set up RE to send documents to db, as the
    # IPython profile (profile_collection) does, so that data acquisition works.
    from bluesky import RunEngine
    RE = RunEngine({})
    from ophyd.sim import det
    from bluesky.plans import count
    RE.subscribe(db.insert)


    # Take some simulated data.
    RE(count([det]), sample='A', purpose='dan is testing for Lutz')


    # This is the part you have to add to make data access work.
    def fetch_sample(start, stop):
    if 'sample' in start:
    query = dict(name=start['sample'])
    return samples.find_one(query)
    # This tells the databroker to addd some custom stuff to the `.ext` attribute
    # of each Header.
    db.external_fetchers['sample_info'] = fetch_sample
    print(db[-1].ext)