API Documents

crystalmapping.plans

exception crystalmapping.plans.TomoPlanError

Error in the plans.

crystalmapping.plans.configure_area_det(detector, exposure, acq_time)

Configure exposure time of a detector in continuous acquisition mode.

crystalmapping.plans.dark_plan(detector)

Take a dark scan in “dark” stream.

crystalmapping.plans.fly_scan_nd(detectors: list, *args, move_velocity: float, time_per_point: float, time_per_frame: float, shutter: object, shutter_open: Any, shutter_close: Any, shutter_wait_open: float = 0.0, shutter_wait_close: float = 0.0, take_dark: bool = True, md: Optional[dict] = None, backoff: float = 0.0, snake: bool = False, open_and_close: bool = False) → Generator

Move on a grid and do a fly scan at each point in the grid.

For example, fly_scan_nd([detector], motor_y, 0, 10, 11, motor_x, 0, 20, 21, motor_fly, 0, 5, 6, time_per_point=10, time_per_frame=1, shutter=shutter, shutter_open=1, shutter_close=0, shutter_wait_open=2, shutter_wait_close=5, move_velocity=5, take_dark=True, md={“task”: “fly scan sample 1”, backoff=0.5, snake=False}) means that set detector so that it will collect one image for 10 s one image contains 10 frames and each frame for 1 s

for y in 0, 1, 2, …, 10: for x in 0, 1, 2, …, 20: move to (x, y) wait 5 s collect dark image during the movement open shutter wait 2 s fly scan the motor_fly from -0.5 to 5.5 collect 6 images during the fly close shutter

Parameters

detectorslist

A list of detectors. The first one must be an area detector. The list shouldn’t include the motors in the args.

*args

patterned like (motor1, start1, stop1, num1,

motor2, start2, stop2, num2, motor3, start3, stop3, num3, … motorN, startN, stopN, numN)

The first motor is the “slowest”, the outer loop. The last motor is the “fly” motor, the non-stoping scan along an axis. For all motors except the first motor, there is a “snake” argument: a boolean indicating whether to following snake-like, winding trajectory or a simple left-to-right trajectory.

move_velocityfloat

The speed for the motors to move to the next grid point.

time_per_pointfloat

The time to collect one image at one point.

time_per_framefloat

The time to collect one frame in a image. One image contains serveral frames.

shutterobject

The fast shutter.

shutter_opentyping.Any

The value of the shutter in open state.

shutter_closetyping.Any

The value of the shutter in close state.

shutter_wait_openfloat, optional

The time between the shutter open and the start of the light image collection, by default 0.

shutter_wait_closefloat, optional

The time between the shutter close and the start of the dark image collection, by default 0.

take_darkbool, optional

If true, take a dark image at the end of the fly scan, by default True

mddict, optional

The dictionary of the metadata to added into the plan, by default None

backofffloat, optional

If non-zero, fly scan from start - backoff to end + backoff, by default 0.

snakebool, optional

If true, snake the axis of the fly scan, by default False

open_and_closebool, optional

If true, open and close shutter at every point, by default False

Yields

Iterator[typing.Generic]

The messages of the plan.

Raises

TomoPlanError

Empty detector list.

TomoPlanError

Not enough motors.

TomoPlanError

Wrong motor positions format.

crystalmapping.plans.grid_scan_nd(detectors: list, *args, snake: Optional[Union[list, bool]] = None, time_per_point: float, time_per_frame: float, shutter: object, shutter_open: Any, shutter_close: Any, shutter_wait_open: float = 0.0, shutter_wait_close: float = 0.0, take_dark: bool = True, md=None) → Generator

Scan over a mesh; each motor is on an independent trajectory.

Parameters

detectorslist

A list of ‘readable’ objects

*args

patterned like (motor1, start1, stop1, num1,

motor2, start2, stop2, num2, motor3, start3, stop3, num3, … motorN, startN, stopN, numN)

The first motor is the “slowest”, the outer loop. The last motor is the “fly” motor, the non-stoping scan along an axis. For all motors except the first motor, there is a “snake” argument: a boolean indicating whether to following snake-like, winding trajectory or a simple left-to-right trajectory.

snakebool, optional

If true, snake the axis of the fly scan, by default None

time_per_pointfloat

The time to collect one image at one point.

time_per_framefloat

The time to collect one frame in a image. One image contains serveral frames.

time_per_pointfloat

The time to collect one image at one point.

time_per_framefloat

The time to collect one frame in a image. One image contains serveral frames.

shutterobject

The fast shutter.

shutter_opentyping.Any

The value of the shutter in open state.

shutter_closetyping.Any

The value of the shutter in close state.

shutter_wait_openfloat, optional

The time between the shutter open and the start of the light image collection, by default 0.

shutter_wait_closefloat, optional

The time between the shutter close and the start of the dark image collection, by default 0.

take_darkbool, optional

If true, take a dark image at the end of the fly scan, by default True

md[type], optional

The dictionary of the metadata to added into the plan, by default None, by default None

Yields

Iterator[typing.Generic]

The messages of the plan.

Raises

TomoPlanError

Empty detector list.

TomoPlanError

Not enough motors.

TomoPlanError

Wrong motor positions format.

crystalmapping.plans.loop_forever(motor: Device, left: float, right: float, timeout: float = inf) → Generator[Any, None, None]

Move motor from left to right and right to left repeatedly until t seconds pass

crystalmapping.servers

class crystalmapping.servers.Commands

A collection of commands.

create_best_effort_config(cfg_file: str) → None

Create the configuration file for best efforts server.

Parameters

cfg_filestr

The configuration file. It is an .ini file.

create_extremum_config(cfg_file: str) → None

Create the configuration file for extremum server.

Parameters

cfg_filestr

The configuration file. It is an .ini file.

run_best_effort(cfg_file: str, test: bool = False) → None

Run best effort server.

Parameters

cfg_filestr

The configuration file. It is an .ini file.

testbool, optional

If True, it is a pytest mode, by default False.

run_extremum(cfg_file: str, test: bool = False) → None

Run extremum server.

Parameters

cfg_filestr

The configuration file. It is an .ini file.

testbool, optional

If True, it is a pytest mode, by default False.

class crystalmapping.servers.Extremum(config: ExtremumConfig)

crystalmapping.crystalmapper

class crystalmapping.crystalmapper.CrystalMapper(_config: Optional[MapperConfig] = None, _dataset: Optional[Dataset] = None, _frames_arr: Optional[DataArray] = None, _dark: Optional[ndarray] = None, _light: Optional[ndarray] = None, _windows: Optional[DataFrame] = None, _intensity: Optional[ndarray] = None, _bkg_intensity: Optional[ndarray] = None, _coords: Optional[List[ndarray]] = None, _metadata: Optional[dict] = None)

The Calculator of the crystal maps.

auto_process() → None

Automatically process the data in the standard protocol.

create_crystal_maps() → None

Create the crystal maps after finding the Bragg spots.

find_bragg_spots() → None

Find the Bragg spots before creating the crystal maps.

load_bluesky_v1(run: Any) → None

Load the data and metadata from the version 1 databroker.

Parameters

runBlueskyRun

A version 1 BlueskyRun (Header).

load_bluesky_v2(run: Any) → None

Load the data and metadata from the version 2 databroker.

Parameters

runBlueskyRun

A version 2 BlueskyRun.

show_dark(*args, **kwargs) → FacetGrid

Show the dark image.

show_frame(index: int, *args, **kwargs) → FacetGrid

Show the frame at that index.

show_intensity(**kwargs) → FacetGrid

Show the intensity array.

show_light(*args, **kwargs) → FacetGrid

Show the light image.

show_light_sub_dark(*args, **kwargs) → FacetGrid

Show the dark subtracted light image.

show_windows(*args, **kwargs) → FacetGrid

Show the windows on the dark subtracted light image.

show_windows_on_frame(index: int, *args, **kwargs) → FacetGrid

Show the windows on the frame at the index.

tune_RoI(number: int, half_width: int) → None

Tune the RoI number and half width.

Parameters

numberint

Number of the RoI regions. Choose from the strongest.

half_widthint

Number of pixels excluding the center for the half width of the RoI region.

visualize(peaks: Optional[List[int]] = None, **kwargs) → None

Show the crystal maps of certain peaks.

Parameters

peakstyping.List[int]

A list of integer of the peaks.

class crystalmapping.crystalmapper.MapperConfig(enable_tqdm: bool = True, image_data_key: str = 'dexela_image', RoI_number: int = 100, RoI_half_width: int = 25, trackpy_kernel_size: int = 25, slice_of_frames: Optional[slice] = None)

exception crystalmapping.crystalmapper.MapperError

crystalmapping.callbacks

class crystalmapping.callbacks.DataFrameDumper(db: Broker)

Dump the dataframe to the database using databroker.

dump_df(df: DataFrame, metadata: Optional[dict] = None)

Dump the data frame into the database with the metadata.

class crystalmapping.callbacks.ImageProcessor(data_key: str, subtrahend: ndarray)

A callback to average frames of images, subtract it by another image, and emit the document.

event(doc, **kwargs)

Receive an event document from the raw stream.

This should be reimplemented by a subclass.

Parameters

docevent

kwargs:

All keyword arguments are passed to process_event()

start(doc, _md=None)

Receive a raw start document, re-emit it for the modified stream

class crystalmapping.callbacks.PeakTracker(data_key: str, diameter: Union[int, tuple], **kwargs)

Track the peaks on a series of images and summarize their position and intensity in a dataframe.

event(doc, **kwargs)

Receive an event document from the raw stream.

This should be reimplemented by a subclass.

Parameters

docevent

kwargs:

All keyword arguments are passed to process_event()

start(doc, _md=None)

Receive a raw start document, re-emit it for the modified stream

class crystalmapping.callbacks.TrackLinker(*, db: Optional[Broker] = None, search_range: Union[float, tuple], **kwargs)

Track the peaks in frame and link them in trajectories.

When a stop is received, the data will be pulled from the databroker and processed. Then, the dataframe will be emitted row by row.

event(doc, **kwargs)

Receive an event document from the raw stream.

This should be reimplemented by a subclass.

Parameters

docevent

kwargs:

All keyword arguments are passed to process_event()

start(doc, _md=None)

Receive a raw start document, re-emit it for the modified stream

stop(doc, _md=None)

Receive a raw stop document, re-emit it for the modified stream

crystalmapping.callbacks.gen_processed_images(images: Iterable[Union[list, ndarray]], subtrahend: Union[list, ndarray]) → Generator[ndarray, None, None]

Generate processed image from a series of images.

The process procedure is (1) turn ot numpy array, (2) average the frames to a two dimensional image, (3) subtract the image and fill zero in negative pixels.

Parameters

images

A iterable of images. The dimensions of each image is no less than 2.

subtrahend

The subtrahend image. The dimensions of each image is no less than 2.

Yields

processed_image

A two dimensional image.

crystalmapping.callbacks.get_dataframe(run: Header) → DataFrame

Get the dataframe from the stream. Drop the time column.

crystalmapping.callbacks.get_mean_frame(frames: ndarray) → ndarray

Average the frames to a two dimensional image.

crystalmapping.callbacks.subtract_image(minuend: ndarray, subtrahend: ndarray) → ndarray

Subtract the image and fill zero in negative pixels.

crystalmapping.peakindexer

class crystalmapping.peakindexer.AngleComparsion(h1: numpy.ndarray, h2: numpy.ndarray, angle_sample: float, angle_grain: float, diff_angle: float)

class crystalmapping.peakindexer.IndexResult(peak1: int, peak2: int, u_mat: ndarray, hkls: ndarray, losses: ndarray, loss: ndarray, ac: AngleComparsion)

The result of peak indexing.

class crystalmapping.peakindexer.IndexerConfig(dspacing_bounds: Optional[Tuple[float, float]] = None, index_agl_tolerance: float = 1.0, index_tth_tolerance: float = 0.1, index_best_n: int = 256, index_all_peaks: bool = True)

Configuration for the PeakIndexer.

exception crystalmapping.peakindexer.IndexerError

Error of the PeakIndexer.

class crystalmapping.peakindexer.PeakIndexer(config: ~crystalmapping.peakindexer.IndexerConfig, _datasets: ~typing.List[~xarray.core.dataset.Dataset] = <factory>, _ai: ~typing.Optional[~pyFAI.azimuthalIntegrator.AzimuthalIntegrator] = None, _cell: ~typing.Optional[~pyFAI.calibrant.Cell] = None, _peak_index: ~typing.Optional[~xarray.core.dataset.Dataset] = None, _peaks: ~typing.Optional[~pandas.core.frame.DataFrame] = None, _ubmatrix: ~crystalmapping.ubmatrix.UBMatrix = <crystalmapping.ubmatrix.UBMatrix object>, _pred: ~crystalmapping.peakindexer.PredictedReflection = PredictedReflection(q=None, d=None, tth=None, hkls=None), _previous_result: ~typing.Optional[~xarray.core.dataset.Dataset] = None)

The Calculator of the crystal maps.

guess_miller_index(peak_ids: List[str]) → None

Guess the index of the peaks in one grain.

Parameters

peakstyping.List[str]

The index of the peaks in the table.

hist_error(peak_ids: Optional[List[str]] = None, size: float = 4.0, bins: Any = 'auto') → None

Plot the histogram of erros.

Parameters

peak_idsList[str], optional

A list of peak IDs to plot, by default None, plot all peaks.

sizefloat, optional

Size in inches for the individual panel, by default 4.

index_peaks_by_U(U_index: List[int]) → None

Index the peaks using the U matrix from previous results.

Parameters

U_indexT.List[int]

0-index of the U matrix to use. It is from best to worst in previous result.

load(data_files: List[str], euler_angles: List[Tuple[float, float, float]], poni_file: str, stru_file: str) → None

Load the necessary data.

Parameters

data_filesList[str]

A list of netcdf4 files output by CrystalMapper.

euler_anglesList[EulerAngle]

A list of (alpha, beta, gamma) value of Euler anlge in ZXY convention.

poni_filestr

A pyFAI poni file containing the geometry information of the experiment.

stru_filestr

A cif file containing the lattice information of the sample.

load_miller_index(filename: str) → None

Load the guessing of Miller index.

Parameters

filenamestr

A netcdf4 file of the guessing results.

load_previous_result(result_file: str) → None

Load the result from the previous peak indexing.

Parameters

result_filestr

Output file from the PeakIndexer.

save_miller_index(filename: str) → None

Save the guess of Miller index.

Parameters

filenamestr

A destination of netcdf4 file.

show(best_n: Optional[int] = None) → None

Print out the indexing results.

Parameters

best_nT.Optional[int], optional

Only print out best n results, by default None

visualize(dataset_id: int, peak_ids: Optional[List[str]] = None, **kwargs) → None

Visualize the crystal maps.

Parameters

dataset_idint

A 0-index ID of the dataset.

peaksList[str], optional

A list of peak id in that dataset, by default None

class crystalmapping.peakindexer.PredictedReflection(q: Optional[ndarray] = None, d: Optional[ndarray] = None, tth: Optional[ndarray] = None, hkls: Optional[List[ndarray]] = None)

The results of the predicted reflection.