contrast.recorders package

Module contents

Provides the Recorder base class as well as recorder subclasses for writing, streaming and plotting data.

Submodules

contrast.recorders.Recorder module

class contrast.recorders.Recorder.Process(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)

Bases: SpawnProcess

Dummy for cleaning up the inheritance documentation.

class contrast.recorders.Recorder.RecorderHeader(scannr, path, snapshot=None, description=None, status=None, **kwargs)

Bases: dict

Helper class to define a specific dict format to send recorders when a new scan starts.

class contrast.recorders.Recorder.RecorderFooter(scannr, path, snapshot=None, description=None, status=None)

Bases: dict

Helper class to define a specific dict format to send recorders when a new scan finishes.

class contrast.recorders.Recorder.Recorder(delay=0.1, **kwargs)

Bases: Gadget, Process

Base class for Recorders. Provides the multiprocessing and queuing functionality.

Parameters:

• delay (float) – Sleep time for the queue checking loop.

• **kwargs – Passed on to base class constructor

run()

The main loop of the recorder.

init()

Override this. Use this method to Initialize the recorder (open windows, open files, etc.).

act_on_header(dct)

Override this. Performs an action when a new scan is started. The key-value pairs of dct are defined by RecorderHeader.

act_on_data(dct)

Override this. Performs an action when a new data package is received. The keys of dct are detector and motor names, the values are their readings at one particular point in the scan.

act_on_footer(dct)

Override this. Performs an action when a scan ends.

periodic_check()

A function which gets called on every iteration of the recorder. Useful for example for checking if files should be closed or whether a plot window still exists.

stop()

Stop a started subprocess safely by putting a poison pill in its queue.

class contrast.recorders.Recorder.DummyRecorder(delay=0.1, **kwargs)

Bases: Recorder

Dummy recorder for practise.

Parameters:

• delay (float) – Sleep time for the queue checking loop.

• **kwargs – Passed on to base class constructor

contrast.recorders.Recorder.active_recorders()

Utilify function which returns a list of currently running Recorder objects.

class contrast.recorders.Recorder.LsRec

Bases: object

This class generates the macro `lsrec`

List active recorders.

contrast.recorders.Hdf5Recorder module

class contrast.recorders.Hdf5Recorder.Link(*args, universal=False, **kwargs)

Bases: ExternalLink

Helper class which wraps a h5py.ExternalLink, but which also informs the Hdf5Recorder about whether there will be one link per scan (universal=True) or one link per position (universal=False).

class contrast.recorders.Hdf5Recorder.Hdf5Recorder(name=None)

Bases: Recorder

Recorder which writes to hdf5 files.

act_on_header(dct)

Opens a file when a new scan starts.

act_on_data(dct, base='entry/measurement/')

Write data packets to the h5 file.

act_on_footer(dct)

Takes another snapshot (post scan) and then closes the file after the scan.

contrast.recorders.PlotRecorder module

contrast.recorders.PlotRecorder.dict_lookup(dct, path)

Helper to recursively get dct[‘path’][‘to’][‘item’] from dct[‘path/to/item’].

class contrast.recorders.PlotRecorder.PlotRecorder(data1, data2=None, name='plot')

Bases: Recorder

Recorder which catches data and plots it with matplotlib.

Unlike the base class Recorder, the GUI event loop takes care of the timing (plt.timer) and when to close (plt.show).

class contrast.recorders.PlotRecorder.LivePlot(data1, data2=None)

Bases: object

This class generates the macro `liveplot`

Start a live plot recorder which will plot coming scans.

liveplot [<x>] <y>

Examples:

liveplot xmotor diode1
liveplot diode1

contrast.recorders.StreamRecorder module

contrast.recorders.StreamRecorder.walk_dict(dct)

A recursive version of dict.items(), which yields (containing-dict, key, val).

class contrast.recorders.StreamRecorder.StreamRecorder(name=None, port=5556)

Bases: Recorder

Recorder which publishes data to a zmq stream. Try receiving it with:

import zmq
context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.connect ("tcp://localhost:5556")
socket.setsockopt(zmq.SUBSCRIBE, b"") # subscribe to all topics
while True:
    messagedata = socket.recv_pyobj()
    print(messagedata)

The result is a sequence of python objects, of type dict or OrderedDict. Each dict contains a 'status' field with one of the following values.

• heartbeat: a dummy message sent to keep connections alive

• started: indicates the beginning of a scan

• running: indicates that this is a data message from an ongoing scan

• finished or interrupted: indicates that a scan was either completed or stopped.

Data messages simply contain all gathered motor and detector data, as placed in the Recorder queue. For example:

$ print(dict(socket.recv_pyobj()))

{'sx': 0.8,
 'det2': <ExternalLink to "entry/measurement/data" in file
         "/tmp/Dummy_scan_005_image_004.hdf5",
 'det1': 0.024625569499185596,
 'dt': 2.3689677715301514,
 'status': 'running'}

zmq = <module 'zmq' from '/home/docs/checkouts/readthedocs.org/user_builds/contrast/envs/latest/lib/python3.14/site-packages/zmq/__init__.py'>

act_on_header(dct)

Relay information.

Converts RecorderHeader to plain dict so the receiver doesn’t need the contrast library.

act_on_data(dct, base='entry/measurement/')

Relay information, but filter out exotic objects like Links.

act_on_footer(dct)

Relay information.