CHAP.tomo package

PipelineItems unique to tomography data processing workflows.

This module contains all the PipelineItems (Processors, Readers and Writers) that are unique to the tomography workflow. Any of these PipelineItems can be used as items in a CHAP Pipeline or instantiated from a user Python script.

Note

Using the tomography workflow pipeline items in a CHAP Pipeline and running it, requires a Tomo conda environent or access to the appropriate CHAP Tomo executable, see Tomography subpackage (CHAP.tomo)

Submodules summary

models

Pydantic model configuration classes unique to the tomography workflow.

processor

Processors unique to the tomography workflow.

reader

Readers unique to the tomography workflow.

writer

Writers unique to the tomography workflow.

Submodules

CHAP.tomo.models module

Pydantic model configuration classes unique to the the tomography workflow.

class Detector

Bases: CHAPBaseModel

Properties of the tomography image detector.

Detector properties of the detector used in the tomography experiment. The image origin is assumed to be in the top-left corner, with rows down (\(-z\) in lab frame) and columns sideways (\(+x\) in lab frame).

Variables:

• prefix (str) – Prefix of the detector in the SPEC file.

• rows (int) – Number of pixel rows on the detector.

• columns (int) – Number of pixel columns on the detector.

• pixel_size (int or list[int]) – Pixel size of the detector in \(mm\).

• lens_magnification (float, optional) – Lens magnification for the detector, defaults to 1.0.

columns: Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), None]

lens_magnification: Annotated[float, None, Interval(gt=0, ge=None, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)] | None

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

pixel_size: Annotated[list[Annotated[float, None, Interval(gt=0, ge=None, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)]], Len(min_length=1, max_length=2)]

prefix: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None, ascii_only=None)]

rows: Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), None]

class TomoCombineConfig(*, x_bounds: Annotated[list[Annotated[int, None, Interval(gt=None, ge=-1, lt=None, le=None), None]], Len(min_length=2, max_length=2)] | None = None, y_bounds: Annotated[list[Annotated[int, None, Interval(gt=None, ge=-1, lt=None, le=None), None]], Len(min_length=2, max_length=2)] | None = None, z_bounds: Annotated[list[Annotated[int, None, Interval(gt=None, ge=-1, lt=None, le=None), None]], Len(min_length=2, max_length=2)] | None = None)

Bases: CHAPBaseModel

Configuration for the combine tomography stacks processor TomoCombineProcessor.

Variables:

• x_bounds (list[int], optional) – Combined image bounds in the x-direction.

• y_bounds (list[int], optional) – Combined image bounds in the y-direction.

• z_bounds (list[int], optional) – Combined image bounds in the z-direction.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

x_bounds: Annotated[list[Annotated[int, None, Interval(gt=None, ge=-1, lt=None, le=None), None]], Len(min_length=2, max_length=2)] | None

y_bounds: Annotated[list[Annotated[int, None, Interval(gt=None, ge=-1, lt=None, le=None), None]], Len(min_length=2, max_length=2)] | None

z_bounds: Annotated[list[Annotated[int, None, Interval(gt=None, ge=-1, lt=None, le=None), None]], Len(min_length=2, max_length=2)] | None

class TomoFindCenterConfig(*, center_stack_index: Annotated[int, None, Interval(gt=None, ge=0, lt=None, le=None), None] | None = None, center_rows: Annotated[list[Annotated[int, None, Interval(gt=None, ge=0, lt=None, le=None), None]], Len(min_length=1, max_length=2)] | None = None, center_offsets: Annotated[list[Annotated[float, None, Interval(gt=None, ge=None, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)]], Len(min_length=1, max_length=2)] | None = None, center_offset_min: Annotated[float, None, Interval(gt=None, ge=None, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)] | None = None, center_offset_max: Annotated[float, None, Interval(gt=None, ge=None, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)] | None = None, center_search_range: Annotated[list[Annotated[float, None, Interval(gt=None, ge=None, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)]], Len(min_length=1, max_length=3)] | None = None, gaussian_sigma: Annotated[float, None, Interval(gt=None, ge=0.0, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)] | None = None, ring_width: Annotated[float, None, Interval(gt=None, ge=0.0, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)] | None = None)

Bases: CHAPBaseModel

Configuration for the tomography center axis finding processor TomoFindCenterProcessor.

Variables:

• center_stack_index (int, optional) – Stack index of the tomography set to find the center axis.

• center_rows (list[int], optional) – Detector image row indices for the center finding processor.

• center_offsets (list[float], optional) – Centers at the center finding row indices in pixels, relative to the center of a detector row.

• center_offset_min (float, optional) – Minimum value of center_offset in center axis finding search in pixels.

• center_offset_max (float, optional) – Maximum value of center_offset in center axis finding search in pixels.

• gaussian_sigma (float, optional) – Standard deviation for the Gaussian filter applied to image reconstruction visualizations, defaults to no filtering performed.

• ring_width (float, optional) – Maximum ring width in pixels used to filter ring artifacts during image reconstruction, defaults to no ring removal performed.

center_offset_max: Annotated[float, None, Interval(gt=None, ge=None, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)] | None

center_offset_min: Annotated[float, None, Interval(gt=None, ge=None, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)] | None

center_offsets: Annotated[list[Annotated[float, None, Interval(gt=None, ge=None, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)]], Len(min_length=1, max_length=2)] | None

center_rows: Annotated[list[Annotated[int, None, Interval(gt=None, ge=0, lt=None, le=None), None]], Len(min_length=1, max_length=2)] | None

center_search_range: Annotated[list[Annotated[float, None, Interval(gt=None, ge=None, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)]], Len(min_length=1, max_length=3)] | None

center_stack_index: Annotated[int, None, Interval(gt=None, ge=0, lt=None, le=None), None] | None

gaussian_sigma: Annotated[float, None, Interval(gt=None, ge=0.0, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)] | None

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

ring_width: Annotated[float, None, Interval(gt=None, ge=0.0, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)] | None

class TomoReconstructConfig(*, x_bounds: Annotated[list[Annotated[int, None, Interval(gt=None, ge=-1, lt=None, le=None), None]], Len(min_length=2, max_length=2)] | None = None, y_bounds: Annotated[list[Annotated[int, None, Interval(gt=None, ge=-1, lt=None, le=None), None]], Len(min_length=2, max_length=2)] | None = None, z_bounds: Annotated[list[Annotated[int, None, Interval(gt=None, ge=-1, lt=None, le=None), None]], Len(min_length=2, max_length=2)] | None = None, secondary_iters: Annotated[int, None, Interval(gt=None, ge=0, lt=None, le=None), None] = 0, gaussian_sigma: Annotated[float, None, Interval(gt=None, ge=0.0, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)] | None = None, remove_stripe_sigma: Annotated[float, None, Interval(gt=None, ge=0.0, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)] | None = None, ring_width: Annotated[float, None, Interval(gt=None, ge=0.0, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)] | None = None)

Bases: CHAPBaseModel

Configuration for the tomography image reconstruction processor TomoReconstructProcessor.

Variables:

• x_bounds (list[int], optional) – Reconstructed image bounds in the x-direction.

• y_bounds (list[int], optional) – Reconstructed image bounds in the y-direction.

• z_bounds (list[int], optional) – Reconstructed image bounds in the z-direction.

• secondary_iters (int, optional) – Number of secondary iterations in the tomopy image reconstruction algorithm, defaults to 0.

• gaussian_sigma (float, optional) – Standard deviation for the Gaussian filter applied to image reconstruction visualizations, defaults to no filtering performed.

• ring_width (float, optional) – Maximum ring width in pixels used to filter ring artifacts during image reconstruction, defaults to no ring removal performed.

gaussian_sigma: Annotated[float, None, Interval(gt=None, ge=0.0, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)] | None

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

remove_stripe_sigma: Annotated[float, None, Interval(gt=None, ge=0.0, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)] | None

ring_width: Annotated[float, None, Interval(gt=None, ge=0.0, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)] | None

secondary_iters: Annotated[int, None, Interval(gt=None, ge=0, lt=None, le=None), None]

x_bounds: Annotated[list[Annotated[int, None, Interval(gt=None, ge=-1, lt=None, le=None), None]], Len(min_length=2, max_length=2)] | None

y_bounds: Annotated[list[Annotated[int, None, Interval(gt=None, ge=-1, lt=None, le=None), None]], Len(min_length=2, max_length=2)] | None

z_bounds: Annotated[list[Annotated[int, None, Interval(gt=None, ge=-1, lt=None, le=None), None]], Len(min_length=2, max_length=2)] | None

class TomoReduceConfig(*, img_row_bounds: Annotated[list[Annotated[int, None, Interval(gt=None, ge=-1, lt=None, le=None), None]], Len(min_length=2, max_length=2)] | None = None, delta_theta: Annotated[float, None, Interval(gt=0, ge=None, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)] | None = None, remove_stripe: dict | None = {})

Bases: CHAPBaseModel

Configuration for the tomography image reduction processor TomoReduceProcessor.

Variables:

• img_row_bounds (list[int], optional) – Detector image bounds in the row-direction (ignored for id1a3 and id3a for an image stack).

• delta_theta (float, optional) – Rotation angle increment in image reduction in degrees.

• remove_stripe (dict(str, dict)) –

  The type(s) of tomopy stripe removal types to apply during data reduction with a dictionary of any additional function parameters.

  Example usage:

  remove_stripe={'remove_all_stripe': {'ncore': 16}}
  

delta_theta: Annotated[float, None, Interval(gt=0, ge=None, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)] | None

img_row_bounds: Annotated[list[Annotated[int, None, Interval(gt=None, ge=-1, lt=None, le=None), None]], Len(min_length=2, max_length=2)] | None

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

remove_stripe: dict | None

class TomoSimConfig

Bases: CHAPBaseModel

Configuration for the tomography simulator processor TomoSimFieldProcessor.

Variables:

• station (Literal['id1a3', 'id3a', 'id3b']) – The station name (in ‘idxx’ format).

• detector (Detector) – Detector configuration in the tomography experiment.

• sample_type (Literal['square_rod', 'square_pipe', 'hollow_cube', 'hollow_brick', 'hollow_pyramid']) – Sample type for the tomography simulator.

• sample_size (list[float]) – Size of each sample dimension in \(mm\) (internally converted to an integer number of pixels). Enter three values for sample_type == ‘hollow_pyramid’, the height and the side at the respective bottom and the top of the pyramid.

• wall_thickness (float) – Wall thickness for pipe, cube, and brick in \(mm\) (internally converted to an integer number of pixels).

• mu (float, optional) – Linear attenuation coefficient in \(mm^{-1}\), defaults to 0.05.

• theta_step (float) – Rotation angle increment in the tomography simulation in degrees.

• beam_intensity (float, optional) – Initial beam intensity in counts, defaults to 1.e9.

• background_intensity (float, optional) – Background intensity in counts, defaults to 20.

• slit_size (float, optional) – Vertical beam height in \(mm\), defaults to 1.0.

background_intensity: Annotated[float, None, Interval(gt=0, ge=None, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)] | None

beam_intensity: Annotated[float, None, Interval(gt=0, ge=None, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)] | None

detector: Detector

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

mu: Annotated[float, None, Interval(gt=0, ge=None, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)] | None

sample_size: Annotated[list[Annotated[float, None, Interval(gt=0, ge=None, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)]], Len(min_length=1, max_length=3)]

sample_type: Literal['square_rod', 'square_pipe', 'hollow_cube', 'hollow_brick', 'hollow_pyramid']

slit_size: Annotated[float, None, Interval(gt=0, ge=None, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)] | None

station: Literal['id1a3', 'id3a', 'id3b']

theta_step: Annotated[float, None, Interval(gt=0, ge=None, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)]

wall_thickness: Annotated[float, None, Interval(gt=None, ge=0.0, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)] | None

CHAP.tomo.processor module

Module for Processors unique to the tomography workflow.

Tomographic reconstruction refers to the process of recovering 3D spatial information on an object from a set of projected images acquired under different angles after transmission of an x-ray beam through the sample. This module contains the CHAP processors that perform the steps in a typical tomographic reconstruction workflow. It also contains CHAP processors to create simulated 3D image data to test the workflow.

NUM_CORE_TOMOPY_LIMIT = 24

int: Maximum number of cores in Tomopy routines.

class SetNumexprThreads(num_proc)

Bases: object

Class that sets and keeps track of the number of processors used by the code in general and by the numexpr package specifically.

class TomoBrightFieldProcessor(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO')

Bases: Processor

A processor to create the bright field associated with a simulated tomography data set created by TomoSimProcessor.

Variables:

num_image (int, optional.) – Number of bright field images, defaults to 5.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:

self: The BaseModel instance. context: The context.

num_image: Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), None] | None

process(data)

Process the input configuration and return a NeXus style NXroot object with the simulated bright field detector images.

Parameters:

data (list[PipelineData]) – Input data.

Raises:

ValueError – Missing or invalid input or configuration parameter.

Returns:

Simulated bright field images.

Return type:

nexusformat.nexus.NXroot

class TomoCHESSMapConverter(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO')

Bases: Processor

A processor to convert a CHESS style tomography experiment map with dark and bright field configurations to a NeXus style NXtomo input format.

Variables:

nxmemory (int, optional) – Maximum memory usage when reading NeXus files.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:

self: The BaseModel instance. context: The context.

nxmemory: Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), None] | None

process(data)

Process the input map and configuration and return a NeXus style NXroot object based on the NXtomo format.

Parameters:

data (list[PipelineData]) – Input map and configuration for tomographic image reduction/reconstruction.

Raises:

• RuntimeError – Inconsistent thetas among tomography image stacks.

• ValueError – Invalid input or configuration parameter.

Returns:

NeXus style tomography input configuration.

Return type:

nexusformat.nexus.NXroot

class TomoCombineProcessor(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO')

Bases: Processor

A processor to combine a stack of reconstructed images returning a NeXus style NXroot object containing the combined data, an optional list of byte stream representions of Matplotlib figures, and the metadata associated with the data reduction step.

Variables:

• config (dict, optional) – Initialization parameters for an instance of TomoCombineConfig.

• num_proc (int, optional) – Number of processors, defaults to 64.

• nxmemory (int, optional) – Maximum memory usage when reading NeXus files.

• save_figures (bool, optional) – Create Matplotlib figures that can be saved to file downstream in the workflow, defaults to True.

config: TomoCombineConfig | None

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:

self: The BaseModel instance. context: The context.

num_proc: Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), None] | None

nxmemory: Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), None] | None

pipeline_fields: dict

process(data)

Combine the reconstructed tomography stacks.

Parameters:

data (list[PipelineData]) – Input data containing the reconstructed data as a NeXus style NXroot object.

Raises:

ValueError – Invalid input or configuration parameter.

Returns:

Metadata associated with the workflow, a list of byte stream representions of Matplotlib figures, and the result of the data combination.

Return type:

PipelineData, PipelineData, PipelineData

save_figures: bool | None

class TomoDarkFieldProcessor(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO')

Bases: Processor

A processor to create the dark field associated with a simulated tomography data set created by TomoSimProcessor.

Variables:

num_image (int, optional.) – Number of dark field images, defaults to 5.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:

self: The BaseModel instance. context: The context.

num_image: Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), None] | None

process(data)

Process the input configuration and return a NeXus style NXroot object with the simulated dark field detector images.

Parameters:

data (list[PipelineData]) – Input data.

Raises:

ValueError – Missing or invalid input or configuration parameter.

Returns:

Simulated dark field images.

Return type:

nexusformat.nexus.NXroot

class TomoFindCenterGui(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO')

Bases: Processor

A processor that creates and opens a GUI to interactively find and return the calibrated center axis information from a set of reduced tomographic images.

Variables:

• tk_root (tkinter.Tk) – tkinter root window.

• tomo_stacks (numpy.ndarray) – Reduced image data stack(s).

• tbf (numpy.ndarray) – Bright field image data.

• thetas (numpy.ndarray) – Rotation angle of the images in each stack in degrees.

• img_row_bounds ([int, int]) – Detector image bounds in the row-direction.

• img_column_bounds ([int, int]) – Detector image bounds in the column-direction

• center_stack_index (int) – Stack index of the tomography set to find the center axis.

• center_rows (list[int], optional) – Detector image row indices for the center finding processor.

• num_center_rows (int, optional) – Number of rows to find the center at, defaults to the 2 or 1 if the reduced data stack only contains one row.

• num_proc (int, optional) – Number of processors, defaults to 64.

• gaussian_sigma (float, optional) – Standard deviation for the Gaussian filter applied to image reconstruction visualizations, defaults to no filtering performed.

• ring_width (float, optional) – Maximum ring width in pixels used to filter ring artifacts during image reconstruction, defaults to no ring removal performed.

property center_offsets

Return the selected centers at the specified or selected center finding rows.

Type:

list[float]

center_rows: Annotated[list[Annotated[int, None, Interval(gt=None, ge=0, lt=None, le=None), None]], Len(min_length=0, max_length=None)] | None

center_stack_index: Annotated[int, None, Interval(gt=None, ge=0, lt=None, le=None), None]

gaussian_sigma: Annotated[float, None, Interval(gt=None, ge=0.0, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)] | None

img_column_bounds: Annotated[list[Annotated[int, None, Interval(gt=None, ge=0, lt=None, le=None), None]], Len(min_length=2, max_length=2)]

img_row_bounds: Annotated[list[Annotated[int, None, Interval(gt=None, ge=0, lt=None, le=None), None]], Len(min_length=2, max_length=2)]

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:

self: The BaseModel instance. context: The context.

num_center_rows: Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), None] | None

num_proc: Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), None] | None

property recon_planes

Return the reconstructed images at the specified or selected center finding row indices.

Type:

list[numpy.ndarray]

static reconstruct_planes(tomo_planes, center_offset, thetas, *, num_proc=1, gaussian_sigma=None, ring_width=None)

Invert the sinogram for a single or multiple tomography planes using tomopy’s recon routine.

Parameters:

• tomo_planes (numpy.ndarray) – The (set of) sinogram(s).

• center_offset (int, list[int]) – Rotation center axis for the current sinograms in tomo_planes.

• thetas (numpy.ndarray) – Rotation angles in degrees for each sinogram in tomo_planes.

• num_proc (int, optional) – Number of processors, defaults to 1.

• gaussian_sigma (float, optional) – Standard deviation for the Gaussian filter applied to image reconstruction visualizations, defaults to no filtering performed.

• ring_width (float, optional) – Maximum ring width in pixels used to filter ring artifacts during image reconstruction, defaults to no ring removal performed.

Raises:

ValueError – Invalid center_offset input parameter.

Returns:

Reconstructed plane(s)

Return type:

numpy.ndarray

ring_width: Annotated[float, None, Interval(gt=None, ge=0.0, lt=None, le=None), None, AllowInfNan(allow_inf_nan=False)] | None

tbf: ndarray

thetas: ndarray

tk_root: Annotated[Tk, SkipValidation]

tomo_stacks: ndarray

class TomoFindCenterProcessor(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO')

Bases: Processor

A processor to find and return the calibrated center axis information from a set of reduced tomographic images. In addition, it returns an optional list of byte stream representions of Matplotlib figures, and the metadata associated with the center calibration step.

Variables:

• config (dict, optional) – Initialization parameters for an instance of TomoFindCenterConfig.

• num_proc (int, optional) – Number of processors, defaults to 64.

• nxmemory (int, optional) – Maximum memory usage when reading NeXus files.

• save_figures (bool, optional) – Create Matplotlib figures that can be saved to file downstream in the workflow, defaults to True.

config: TomoFindCenterConfig | None

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:

self: The BaseModel instance. context: The context.

num_proc: Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), None] | None

nxmemory: Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), None] | None

pipeline_fields: dict

process(data)

Find the calibrated center axis information

Parameters:

data (list[PipelineData]) – Input data containing the reduced data as a NeXus style NXroot object and optionally a center axis calibration processor configuration.

Raises:

ValueError – Unable to find valid reduced data in input data.

Returns:

Metadata associated with the workflow, a list of byte stream representions of Matplotlib figures, and the calibrated center axis information.

Return type:

PipelineData, PipelineData, PipelineData

save_figures: bool | None

class TomoMetadataProcessor(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO')

Bases: Processor

A processor that takes data from the FOXDEN Data Discovery or Metadata service and extracts what’s available to create a MapConfig instance for a tomography experiment.

Variables:

config (dict) – Configuration dictionary containing all fields required to create a MapConfig instance that are not available from the metadata record.

config: dict

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:

self: The BaseModel instance. context: The context.

process(data)

Process the metadata and return a dictionary with extracted data to create a MapConfig instance for the tomography experiment.

Parameters:

data (list[PipelineData]) – Input data.

Raises:

ValueError – Invalid input or configuration parameter.

Returns:

Metadata from the tomography experiment.

Return type:

MapConfig

class TomoReconstructProcessor(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO')

Bases: Processor

A processor to reconstruct a set of reduced images returning a NeXus style NXroot object containing the reconstructed data, an optional list of byte stream representions of Matplotlib figures, and the metadata associated with the data reduction step.

Variables:

• config (dict, optional) – Initialization parameters for an instance of TomoReconstructConfig.

• center_config (dict, optional) – Center axis calibration configuration.

• num_proc (int, optional) – Number of processors, defaults to 64.

• nxmemory (int, optional) – Maximum memory usage when reading NeXus files.

• save_figures (bool, optional) – Create Matplotlib figures that can be saved to file downstream in the workflow, defaults to True.

center_config: TomoFindCenterConfig | None

config: TomoReconstructConfig | None

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:

self: The BaseModel instance. context: The context.

num_proc: Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), None] | None

nxmemory: Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), None] | None

pipeline_fields: dict

process(data)

Reconstruct the tomography data.

Parameters:

data (list[PipelineData]) – Input data containing the reduced data as a NeXus style NXroot object, the center axis information and optionally a reconstruct data processor configuration.

Raises:

• RuntimeError – Dimension mismatch in center_offsets.

• ValueError – Invalid input or configuration parameter.

Returns:

Metadata associated with the workflow, a list of byte stream representions of Matplotlib figures, and the result of the data reconstruction.

Return type:

PipelineData, PipelineData, PipelineData

save_figures: bool | None

class TomoReduceProcessor(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO')

Bases: Processor

A processor to reduce a set of raw tomographic images returning a NeXus style NXroot object containing the data after correcting the images for the presample intensity (optionally) and normalization with dark and bright field, an optional list of byte stream representions of Matplotlib figures, and the metadata associated with the data reduction step.

Variables:

• config (dict, optional) – Initialization parameters for an instance of TomoReduceConfig.

• num_proc (int, optional) – Number of processors, defaults to 64.

• nxmemory (int, optional) – Maximum memory usage when reading NeXus files.

• save_figures (bool, optional) – Create Matplotlib figures that can be saved to file downstream in the workflow, defaults to True.

config: TomoReduceConfig | None

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:

self: The BaseModel instance. context: The context.

num_proc: Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), None] | None

nxmemory: Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), None] | None

pipeline_fields: dict

process(data)

Reduced the tomography images.

Parameters:

data (list[PipelineData]) – Input data containing the raw data as a NeXus style NXroot object.

Raises:

• RuntimeError – Invalid dark or bring field shape or unable to load valid (the) tomography image stack(s).

• TypeError – Error progagated from numexpr.evaluate().

• ValueError – Invalid input or configuration parameter.

Returns:

Metadata associated with the workflow, a list of byte stream representions of Matplotlib figures, and the result of the data reduction.

Return type:

PipelineData, PipelineData, PipelineData

save_figures: bool | None

class TomoSimFieldProcessor(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO')

Bases: Processor

A processor to create a simulated tomography data set returning a NeXus style NXroot object containing the simulated tomography detector images.

Variables:

config (dict, optional) – Initialization parameters for an instance of TomoSimConfig.

config: TomoSimConfig

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:

self: The BaseModel instance. context: The context.

pipeline_fields: dict

process(data)

Process the input configuration and return a NeXus style NXroot object with the simulated tomography detector images.

Parameters:

data (list[PipelineData]) – Input data.

Raises:

ValueError – Invalid input or configuration parameter.

Returns:

Simulated tomographic images.

Return type:

nexusformat.nexus.NXroot

class TomoSpecProcessor(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO')

Bases: Processor

A processor to create a tomography SPEC file associated with a simulated tomography data set created by TomoSimProcessor.

Variables:

• filename (str, optional) – Metadata input filename, when running with FOXDEN.

• scan_numbers (list[int], optional) – List of SPEC scan numbers.

filename: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None, ascii_only=None)] | None

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:

self: The BaseModel instance. context: The context.

process(data)

Process the input configuration and return a list of strings representing a plain text SPEC file.

Parameters:

data (list[PipelineData]) – Input data.

Raises:

ValueError – Invalid input or configuration parameter.

Returns:

Simulated SPEC file.

Return type:

nexusformat.nexus.NXroot or (PipelineData, PipelineData)

scan_numbers: Annotated[list[Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), None]], Len(min_length=1, max_length=None)] | None

classmethod validate_scan_numbers(scan_numbers)

Validate the specified list of scan numbers.

Parameters:

scan_numbers (int or list[int] or str) – List of scan numbers.

Returns:

Validated scan numbers.

Return type:

list[int]

validate_tomospecprocessor_after()

Validate the TomoSpecProcessor configuration.

Returns:

Validated model configuration

Return type:

TomoSpecProcessor

create_metadata_provenance(did_suffix, data=None, metadata=None, provenance=None, user_metadata=None, logger=None, update=False, read=True)

read_metadata_provenance(data, logger=None, remove=True)

CHAP.tomo.reader module

Module for Readers unique to the tomography workflow.

CHAP.tomo.writer module

Module for Writers unique to the tomography workflow.

class TomoWriter(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO')

Bases: Writer

Writer for saving tomo data.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:

self: The BaseModel instance. context: The context.

validate_tomowriter_after()

Validate the filename extension.

Returns:

Validated writer configuration

Return type:

TomoWriter

write(data)

Write the results of the (partial) tomographic reconstruction and add provenance data to the data pipeline.

Parameters:

data (list[PipelineData]) – Input data.

Returns:

Output data.

Return type:

list[PipelineData]