CHAP.common package

PipelineItems that are or can be used in workflows for processing data from multiple different X-ray techniques.

Any of these PipelineItems can be used as items in a CHAP Pipeline or instantiated from a user Python script.

Submodules summary

map_utils

Common map data model functions and classes.

models

Subpackage containing Pydantic model configuration classes for PipelineItems that are common to various processing workflows.

nexus_utils

PipelineItems for interacting with NeXus file objects.

processor

Module for generic Processors used in multiple experiment-specific workflows.

reader

Module for generic Readers used in multiple experiment-specific workflows.

utils

Some generic utility functions.

writer

Module for generic Writers used in multiple experiment-specific workflows.

Subpackages

• CHAP.common.models package
  • Submodules
  • CHAP.common.models.common module
    • BinarizeConfig
      • BinarizeConfig.method
      • BinarizeConfig.model_config
      • BinarizeConfig.num_bin
      • BinarizeConfig.nxpath
      • BinarizeConfig.remove_original_data
    • ImageProcessorConfig
      • ImageProcessorConfig.animation
      • ImageProcessorConfig.axis
      • ImageProcessorConfig.basename
      • ImageProcessorConfig.coord_range
      • ImageProcessorConfig.fileformat
      • ImageProcessorConfig.index_range
      • ImageProcessorConfig.model_config
      • ImageProcessorConfig.validate_index_range()
      • ImageProcessorConfig.validate_vrange()
      • ImageProcessorConfig.vrange
    • UnstructuredToStructuredConfig
      • UnstructuredToStructuredConfig.model_config
      • UnstructuredToStructuredConfig.nxpath
      • UnstructuredToStructuredConfig.signals
      • UnstructuredToStructuredConfig.unstructured_axes
      • UnstructuredToStructuredConfig.validate_nxpath()
      • UnstructuredToStructuredConfig.validate_signals()
      • UnstructuredToStructuredConfig.validate_unstructured_axes()
  • CHAP.common.models.integration module
    • AzimuthalIntegratorConfig
      • AzimuthalIntegratorConfig.ai
      • AzimuthalIntegratorConfig.mask_data
      • AzimuthalIntegratorConfig.mask_file
      • AzimuthalIntegratorConfig.model_config
      • AzimuthalIntegratorConfig.model_post_init()
      • AzimuthalIntegratorConfig.params
      • AzimuthalIntegratorConfig.poni_file
      • AzimuthalIntegratorConfig.validate_ai()
      • AzimuthalIntegratorConfig.validate_root()
    • Integrate1dConfig
      • Integrate1dConfig.method
      • Integrate1dConfig.model_config
      • Integrate1dConfig.npt
    • Integrate2dConfig
      • Integrate2dConfig.method
      • Integrate2dConfig.model_config
      • Integrate2dConfig.npt_azim
      • Integrate2dConfig.npt_rad
    • IntegrateRadialConfig
      • IntegrateRadialConfig.method
      • IntegrateRadialConfig.model_config
      • IntegrateRadialConfig.npt
      • IntegrateRadialConfig.radial_unit
    • MultiGeometryConfig
      • MultiGeometryConfig.ais
      • MultiGeometryConfig.azimuth_range
      • MultiGeometryConfig.chi_disc
      • MultiGeometryConfig.empty
      • MultiGeometryConfig.model_config
      • MultiGeometryConfig.radial_range
      • MultiGeometryConfig.unit
      • MultiGeometryConfig.validate_ais()
      • MultiGeometryConfig.wavelength
    • PyfaiIntegrationConfig
      • PyfaiIntegrationConfig.azimuthal_integrators
      • PyfaiIntegrationConfig.integrations
      • PyfaiIntegrationConfig.model_config
      • PyfaiIntegrationConfig.sum_axes
      • PyfaiIntegrationConfig.validate_config()
      • PyfaiIntegrationConfig.zarr_tree()
    • PyfaiIntegratorConfig
      • PyfaiIntegratorConfig.get_axes_indices()
      • PyfaiIntegratorConfig.get_placeholder_data()
      • PyfaiIntegratorConfig.init_placeholder_results()
      • PyfaiIntegratorConfig.integrate()
      • PyfaiIntegratorConfig.integration_method
      • PyfaiIntegratorConfig.integration_params
      • PyfaiIntegratorConfig.model_config
      • PyfaiIntegratorConfig.model_post_init()
      • PyfaiIntegratorConfig.multi_geometry
      • PyfaiIntegratorConfig.name
      • PyfaiIntegratorConfig.result_coords
      • PyfaiIntegratorConfig.result_shape
      • PyfaiIntegratorConfig.right_handed
      • PyfaiIntegratorConfig.validate_config()
      • PyfaiIntegratorConfig.zarr_tree()
  • CHAP.common.models.map module
    • CHAPSlice
      • CHAPSlice.end
      • CHAPSlice.model_config
      • CHAPSlice.start
      • CHAPSlice.step
      • CHAPSlice.tolist()
      • CHAPSlice.toslice()
    • CorrectionsData
      • CorrectionsData.data_type
      • CorrectionsData.label
      • CorrectionsData.model_config
      • CorrectionsData.reserved_labels()
    • Detector
      • Detector.attrs
      • Detector.get_id()
      • Detector.id_
      • Detector.model_config
      • Detector.shape
      • Detector.validate_detector_after()
      • Detector.validate_id()
    • DetectorConfig
      • DetectorConfig.detectors
      • DetectorConfig.model_config
      • DetectorConfig.roi
      • DetectorConfig.roitoslice()
      • DetectorConfig.tolist()
      • DetectorConfig.validate_roi()
    • DwellTimeActual
      • DwellTimeActual.label
      • DwellTimeActual.model_config
      • DwellTimeActual.units
    • IndependentDimension
      • IndependentDimension.end
      • IndependentDimension.model_config
      • IndependentDimension.start
      • IndependentDimension.step
    • MapConfig
      • MapConfig.all_scalar_data
      • MapConfig.attrs
      • MapConfig.coords
      • MapConfig.did
      • MapConfig.dims
      • MapConfig.dwell_time_actual
      • MapConfig.experiment_type
      • MapConfig.get_coords()
      • MapConfig.get_detector_data()
      • MapConfig.get_scan_step_index()
      • MapConfig.get_smb_par_attr()
      • MapConfig.get_value()
      • MapConfig.independent_dimensions
      • MapConfig.model_config
      • MapConfig.model_post_init()
      • MapConfig.postsample_intensity
      • MapConfig.presample_intensity
      • MapConfig.sample
      • MapConfig.scalar_data
      • MapConfig.scan_step_indices
      • MapConfig.shape
      • MapConfig.spec_scans
      • MapConfig.station
      • MapConfig.title
      • MapConfig.validate_attrs()
      • MapConfig.validate_before()
      • MapConfig.validate_data_present
      • MapConfig.validate_experiment_type()
      • MapConfig.validate_mapconfig_before()
    • PointByPointScanData
      • PointByPointScanData.data_type
      • PointByPointScanData.get_value()
      • PointByPointScanData.label
      • PointByPointScanData.model_config
      • PointByPointScanData.name
      • PointByPointScanData.ndigits
      • PointByPointScanData.units
      • PointByPointScanData.validate_for_scalar_data()
      • PointByPointScanData.validate_for_spec_scans()
      • PointByPointScanData.validate_for_station()
      • PointByPointScanData.validate_label()
    • PostsampleIntensity
      • PostsampleIntensity.label
      • PostsampleIntensity.model_config
      • PostsampleIntensity.units
    • PresampleIntensity
      • PresampleIntensity.label
      • PresampleIntensity.model_config
      • PresampleIntensity.units
    • Sample
      • Sample.description
      • Sample.model_config
      • Sample.name
    • SpecConfig
      • SpecConfig.experiment_type
      • SpecConfig.model_config
      • SpecConfig.spec_scans
      • SpecConfig.station
      • SpecConfig.validate_experiment_type()
      • SpecConfig.validate_specconfig_before()
    • SpecScans
      • SpecScans.get_detector_data()
      • SpecScans.get_index()
      • SpecScans.get_scanparser()
      • SpecScans.model_config
      • SpecScans.par_file
      • SpecScans.scan_numbers
      • SpecScans.scanparsers
      • SpecScans.spec_file
      • SpecScans.validate_par_file()
      • SpecScans.validate_scan_numbers()
      • SpecScans.validate_spec_file()
    • get_available_scan_numbers()
    • get_detector_data()
    • get_detector_log_timestamps()
    • get_expression_value()
    • get_scan_start_time()
    • get_scanparser()
    • get_smb_par_value()
    • get_spec_counter_value()
    • get_spec_motor_value()
    • import_scanparser()
    • validate_data_source_for_map_config()

Submodules

CHAP.common.map_utils module

Common map data model functions and classes.

class MapSliceProcessor(*, 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

Proccessor for getting partial map data for filling in a NeXus structure created by MapProcessor with fill_data=False. Good for parallelizing workflows across multiple pipelines or processing data live when a scan is still incomplete. Returned data is suitable for writing to an existing map structure with NexusValuesWriter or ZarrValuesWriter.

Variables:

• map_config (MapConfig) – Map configuration.

• detectors (list[Detector]) – Detector configurations.

• spec_file (str) – SPEC file containing scan from which to read a slice of raw data.

• scan_number (int) – Number of scan from which to read a slice of raw data.

detectors: Annotated[list[Detector], Len(min_length=1, max_length=None)]

map_config: MapConfig

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, idx_slice={'start': 0, 'step': 1})

Aggregate partial spec and detector data from one scan in a map, returning results in a format suitable for writing to the full map container with NexusValuesWriter or ZarrValuesWriter.

Parameters:

data (list[PipelineData]) – Result of Reader.read where at least one item has the value ‘common.models.map.MapConfig’ for the ‘schema’ key.

Returns:

Slice of map data, ready to be written to a map container.

Return type:

list[dict[str, Any]]

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

spec_file: Annotated[Path, PathType(path_type=file)]

class SpecScanToMapConfigProcessor(*, 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

Processor to get the MapConfig dictionary configuration representation of a single CHESS SPEC scan.

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, spec_file, scan_number, station, experiment, dwell_time_actual_counter_name, presample_intensity_counter_name, postsample_intensity_counter_name=None, validate_data_present=True)

Return a dictionary representing a valid MapConfig object that contains only the single given scan.

Parameters:

• spec_file (str) – Spec file name

• scan_number (int) – Scan number

• station (Literal["id1a3", "id3a", "id3b", "id4b"]) – Name of the station at which the data was collected.

• experiment – Experiment type

• dwell_time_actual_counter_name (str) – Name of the counter used to record the actual dwell time at time of data collection.

• presample_intensity_counter_name (str) – Name of the counter used to record the incident beam intensity at time of data collection.

• postsample_intensity_counter_name (str, optional) – Name of the counter used to record the post sample beam intensity at time of data collection.

• validate_data_present – Optional validate_data_present key-value pair to the output map configuration, defaults to True.

Returns:

Single-scan map configuration

Return type:

dict

get_axes(nxdata, skip_axes=None)

Get the axes of a NeXus style NXdata object.

Parameters:

• nxdata (nexusformat.nexus.NXdata) – Input data.

• skip_axes (list[str], optional) – Axes to skip.

Returns:

The axes of the NXdata object excluding those in the optional skip_axes parameter.

Return type:

list[str]

CHAP.common.nexus_utils module

PipelineItems for interacting with NeXus file objects.

class NexusMakeLinkProcessor(*, 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

Processor to run makelink within a given NeXus style NXroot object.

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, link_from, link_to, nxname=None, abspath=False)

Create links between Nexus objects within the given PipelineData.

This method takes a NeXus style NXroot object and creates links from the objects specified in link_from to those in link_to. If the underlying file is read-only, a copy is made before modifying. Returns the modified NXroot object containing both the targets and their linked counterparts.

Parameters:

• data (list[PipelineData]) – Input data.

• link_from (str | list[str]) – Path(s) within the NXroot whose objects should be linked. Can be a single path (str) or a list of paths.

• link_to (str | list[str]) – Path(s) within the NXroot that serve as link targets. Can be a single path (str) or a list of paths.

• nxname (str | None, optional) – Name to assign to the created link. If None (default), the default naming rules from makelink are applied.

• abspath (bool, optional) – Whether to create an absolute link path (True) or a relative one (False), defaults to False.

Returns:

The modified NXroot object containing the new links.

Return type:

nexusformat.nexus.NXroot

CHAP.common.processor module

Module for generic Processors used in multiple experiment-specific workflows.

class AsyncProcessor(*, 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 process multiple sets of input data via asyncio module.

Variables:

mgr (Processor) – The Processor used to process every set of input 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.

process(data)

Asynchronously process the input documents with the self.mgr Processor.

Parameters:

data (list[PipelineData]) – Input data.

class BinarizeProcessor(*, 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 binarize a dataset.

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, config=None)

Plot and return a binarized dataset from a dataset contained in data. The dataset must either be array-like or a NeXus style NXobject object with a default plottable data path or a specified path to a NeXus style NXdata or NXfield object.

Parameters:

• data (list[PipelineData]) – Input data.

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

Returns:

Binarized dataset for an array-like input or a return type equal that of the input object with the binarized dataset added.

Return type:

numpy.ndarray | nexusformat.nexus.NXobject

class ConstructBaseline(*, 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 construct a baseline for a dataset.

static construct_baseline(y, x=None, mask=None, tol=1e-06, lam=1000000.0, max_iter=20, title=None, xlabel=None, ylabel=None, interactive=False, return_buf=False)

Construct and return the baseline for a dataset.

Parameters:

• y (numpy.ndarray) – Input data.

• x (array-like, optional) – Independent dimension (only used when interactive is True of when filename is set).

• mask (array-like, optional) – Mask to apply to the spectrum before baseline construction.

• tol (float, optional) – Convergence tolerence, defaults to 1.e-6.

• lam (float, optional) – &lambda (smoothness) parameter (the balance between the residual of the data and the baseline and the smoothness of the baseline). The suggested range is between 100 and 10^8, defaults to 10^6.

• max_iter (int, optional) – Maximum number of iterations, defaults to 20.

• title (str, optional) – Title for the displayed figure.

• xlabel (str, optional) – Label for the x-axis of the displayed figure.

• ylabel (str, optional) – Label for the y-axis of the displayed figure.

• interactive (bool, optional) – Allows for user interactions, defaults to False.

• return_buf (bool, optional) – Return an in-memory object as a byte stream represention of the Matplotlib figure, defaults to False.

Returns:

Smoothed baseline and the configuration and a byte stream represention of the Matplotlib figure if return_buf is True (None otherwise)

Return type:

numpy.ndarray, dict, io.BytesIO | 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, x=None, mask=None, tol=1e-06, lam=1000000.0, max_iter=20, save_figures=False)

Construct and return the baseline for a dataset.

Parameters:

• data (numpy.ndarray | list[PipelineData]) – Input data.

• x (array-like, optional) – Independent dimension (only used when running interactively or when filename is set).

• mask (array-like, optional) – Mask to apply to the spectrum before baseline construction.

• tol (float, optional) – Convergence tolerence, defaults to 1.e-6.

• lam (float, optional) – &lambda (smoothness) parameter (the balance between the residual of the data and the baseline and the smoothness of the baseline). The suggested range is between 100 and 10^8, defaults to 10^6.

• max_iter (int, optional) – Maximum number of iterations, defaults to 20.

• save_figures (bool, optional) – Save .pngs of plots for checking inputs & outputs of this Processor, defaults to False.

Returns:

Smoothed baseline and the configuration.

Return type:

numpy.ndarray, dict

class ConvertStructuredProcessor(*, 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

Processor for converting map data between structured / unstructued formats.

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)

Return the converted map data

Parameters:

data (list[PipelineData]) – Input data.

Returns:

Converted data.

Return type:

nexusformat.nexus.NXdata

class ExpressionProcessor(*, 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

Processor to perform an arbitrary expression on input 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.

process(data, expression, symtable=None, nxprocess=False, nxfieldtable=None, nxdata_name='data', nxfield_name='result')

Return result of plugging input data into the given mathematical expression.

Parameters:

• data (list[PipelineData]) – Input data.

• expression (str) – Mathemetical expression. May use the built-in function round and / or numpy functions with np.<function_name> or numpy.<function_name>.

• symtable (dict[str, (float, int)], optional.) – Values to use for names in expression that should not be obtained from input data. Defaults to None.

• nxprocess (bool, optional) – Flag to indicate the results should be returned as a NeXus style NXprocess defaults to False.

• nxfieldtable (dict[str, nexusformat.nexus.NXfield], optional) – Used only if nxprocess is True. Dictionary of additional NXfield objects to include in the NXprocess, the result object right next to the expression result’s NXfield. Dictionary keys become NXfield names in the returned object. Defaults to None.

• nxdata_name (str, optional) – Used only if nxprocess is True. Name for the NeXus style NXdata object in the returned NXprocess object that contains actual result data. efaults to ‘data’.

• nxfield_name (str, optional) – Used only if nxprocess is True. Name for the NXfield dataset that contains the evaluated expression results. Defaults to ‘result’.

Returns:

Result of evaluating the expression.

Return type:

object

class ImageProcessor(*, 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 perform various visualization operations on images (slices) selected from a NeXus style NXobject.

Variables:

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

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

• save_figures (bool, optional) – Return the plottable image(s) to be written to file downstream in the pipeline, defaults to True.

config: ImageProcessorConfig

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

pipeline_fields: dict

process(data)

Plot and/or return image slices from a NeXus style NXobject. object with a default plottable data path.

Parameters:

data (list[PipelineData]) – Input data.

Returns:

Plottable image(s) (for save_figures = True) or the input default NeXus style NXdata object (for save_figures = False).

Return type:

bytes | nexusformat.nexus.NXdata | numpy.ndarray

save_figures: bool | None

class MPICollectProcessor(*, 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 collects the distributed worker data from MPIMapProcessor on the root node.

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, comm, root_as_worker=True)

Collect data on root node.

Parameters:

• data (list[PipelineData]) – Input data.

• comm (mpi4py.MPI.Comm, optional) – MPI communicator.

• root_as_worker (bool, optional) – Use the root node as a worker, defaults to True.

Returns:

Distributed worker data on the root node.

Return type:

list

class MPIMapProcessor(*, 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 applies a parallel generic sub-pipeline to a map configuration.

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, config=None, sub_pipeline=None)

Run a parallel generic sub-pipeline.

Parameters:

• data (list[PipelineData]) – Input data.

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

• sub_pipeline (Pipeline, optional) – Sub-pipeline.

Returns:

data field of the first item in the returned list of sub-pipeline items.

Return type:

Any

class MPISpawnMapProcessor(*, 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 applies a parallel generic sub-pipeline to a map configuration by spawning workers processes.

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, num_proc=1, root_as_worker=True, collect_on_root=False, sub_pipeline=None)

Spawn workers running a parallel generic sub-pipeline.

Parameters:

• data (list[PipelineData]) – Input data.

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

• root_as_worker (bool, optional) – Use the root node as a worker, defaults to True.

• collect_on_root (bool, optional) – Collect the result of the spawned workers on the root node, defaults to False.

• sub_pipeline (Pipeline, optional) – Sub-pipeline.

Returns:

data field of the first item in the returned list of sub-pipeline items.

class MapProcessor(*, 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 a map configuration and returns a NeXus style NXentry object representing that map’s metadata and any scalar-valued raw data requested by the supplied map configuration.

Variables:

• config (dict | MapConfig) – Map configuration parameters to initialize an instance of MapConfig. Any values in ‘config’ supplant their corresponding values obtained from the pipeline data configuration.

• detector_config (dict | DetectorConfig) – Detector configurations of the detectors to include raw data for in the returned NXentry object (overruling detector info in the pipeline data, if present).

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

config: MapConfig | None

detector_config: DetectorConfig

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

pipeline_fields: dict

process(data, placeholder_data=False, fill_data=True, comm=None)

Process that takes a map configuration and returns a NeXus style NXentry object representing the map.

Parameters:

• data (list[PipelineData]) – Pipeline data list with an optional item for the map configuration parameters with ‘common.models.map.MapConfig’ as its ‘schema’ key.

• placeholder_data (object, optional) – For SMB EDD maps only. Value to use for missing detector data frames, or False if missing data should raise an error, defaults to False.

• fill_data (bool, optional) – Flag to indicate whether or not to fill out datasets with real data; defaults to True.

• comm (mpi4py.MPI.Comm, optional) – MPI communicator.

Returns:

Map data and metadata.

Return type:

Union[nexusformat.nexus.NXentry

classmethod validate_num_proc(num_proc, info)

Validate the number of processors.

Parameters:

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

• info (pydantic.ValidationInfo) – Model parameter validation information.

Returns:

Validated number of processors

Return type:

str

class NXdataToDataPointsProcessor(*, 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

Transform a NeXus style NXdata object into a list of dictionaries. Each dictionary represents a single data point in the coordinate space of the dataset. The keys are the names of the signals and axes in the dataset, and the values are a single scalar value (in the case of axes) or the value of the signal at that point in the coordinate space of the dataset (in the case of signals – this means that values for signals may be any shape, depending on the shape of the signal itself).

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)

Return a list of dictionaries representing the coordinate and signal values at every point in the dataset provided.

Parameters:

data (list[PipelineData]) – Input data.

Returns:

List of all data points in the dataset.

Return type:

list[dict[str,object]]

class NexusToNumpyProcessor(*, 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 the default plottable data in a NeXus style NXobject, object into a numpy.ndarray.

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)

Return the default plottable data signal in a NeXus style NXobject, object contained in data as an numpy.ndarray.

Parameters:

data (list[PipelineData]) – Input data.

Raises:

ValueError – If data has no default plottable data signal.

Returns:

Default plottable data signal.

Return type:

numpy.ndarray

class NexusToXarrayProcessor(*, 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 the default plottable data in a NeXus style NXobject, object into an xarray.DataArray.

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)

Return the default plottable data signal in a NeXus style NXobject, object contained in data as an xarray.DataArray.

Parameters:

data (list[PipelineData]) – Input data.

Raises:

ValueError – If metadata for xarray is absent from data

Returns:

Default plottable data signal.

Return type:

xarray.DataArray

class NexusToZarrProcessor(*, 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

Converter for NeXus to Zarr format.

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, chunks='auto')

Copy and return a Zarr group object from a NeXus style NXgroup object.

Parameters:

data (list[PipelineData]) – Input data.

Returns:

Zarr style group object.

Return type:

zarr.Group

class NormalizeMapProcessor(*, 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

Processor for calling NormalizeNexusProcessor for (usually all) detector data in a NeXus style NXroot object created by MapProcessor

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, normalize_by_nxfield, detector_ids=None)

Return copy of the original input map with additional fields containing normalized detector data.

Parameters:

• data (list[PipelineData]) – Input data.

• normalize_by_nxfield (str) – Path in data to the NXfield containing normalization data.

Returns:

Copy of input data with additional normalized fields.

Return type:

nexusformat.nexus.NXroot

class NormalizeNexusProcessor(*, 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

Processor for scaling one or more NXfield objects in the input NeXus style NXgroup object by the values of another NXfield in the same object .

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, normalize_nxfields, normalize_by_nxfield)

Return copy of the original input NeXus style NXgroup object with additional fields containing the normalized data of each field in normalize_nxfields.

Parameters:

• data (list[PipelineData]) – Input data. to normalize them.

• normalize_nxfields (list[str])

• normalize_by_nxfield (str) – Path in data to the NXfield containing normalization data

Returns:

Copy of input data with additional normalized fields

Return type:

nexusformat.nexus.NXgroup

class NumpyStackProcessor(*, 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

Processor for joining a sequence of arrays along a new axis.

Uses (numpy.stack)[https://numpy.org/doc/stable/reference/generated/numpy.stack.html].

Variables:

• stack_order – List of names of input data arrays to determine order of stacking. If not specified, data arrays are stacked in the exact order input to the Processor. Defaults to None.

• kwargs – Dictionary of keyword arguments to (numpy.stack)[https://numpy.org/doc/stable/reference/generated/numpy.stack.html]; defaults to {}

kwargs: dict | 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)

Extract the contents of the input data, add a string to it, and return the amended value.

Parameters:

data (list[PipelineData]) – Input data.

Returns:

Processed data.

Return type:

str

stack_order: Annotated[list[str], Len(min_length=1, max_length=None)] | None

class NumpySumProcessor(*, 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

Processor for summing an array of elements over a given axis.

Uses (numpy.sum)[https://numpy.org/doc/stable/reference/generated/numpy.sum.html].

Variables:

kwargs – Dictionary of keyword arguments to (numpy.sum)[https://numpy.org/doc/stable/reference/generated/numpy.sum.html]; defaults to {}.

kwargs: dict | 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)

Extract the contents of the input data, add a string to it, and return the amended value.

Parameters:

data (list[PipelineData]) – Input data.

Returns:

Processed data.

Return type:

str

class NumpyToNXfieldProcessor(*, 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

Processor for converting a numpy array into an NXfield.

Variables:

• value – Name of input data array to use as the field’s values. If unspecified, use the last array-like data object in the input data list. Defaults to None.

• kwargs – Dictionary of keyword arguments to [nexusformat.nexus.tree.NXfield](https://nexpy.github.io/nexpy/treeapi.html#nexusformat.nexus.tree.NXfield); defaults to {}

kwargs: dict | 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)

Extract the contents of the input data, add a string to it, and return the amended value.

Parameters:

data (list[PipelineData]) – Input data.

Returns:

Processed data.

Return type:

str

value: str | None

class PandasToXarrayProcessor(*, 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

Converter for pandas.DataFrame to xarray.DataArray or xarray.Dataset

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)

Return input dataframe converted to xarray.

Parameters:

data (list[PipelineData]) – Input data.

Returns:

Input dataframe as xarray.

Return type:

xarray.DataArray | xarray.Dataset

class PrintProcessor(*, 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 simply print the input data to stdout and return the original input data, unchanged in any way.

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)

Print and return the input data.

Parameters:

data (list[PipelineData]) – Input data.

Returns:

data

Return type:

Any

class PyfaiAzimuthalIntegrationProcessor(*, 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

Processor to azimuthally integrate one or more frames of 2d detector data using the pyFAI package.

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, poni_file, npt, mask_file=None, integrate1d_kwargs=None)

Azimuthally integrate the detector data provided and return the result as a dictionary of numpy arrays containing the values of the radial coordinate of the result, the intensities along the radial direction, and the poisson errors for each intensity spectrum.

Parameters:

• data (PipelineData | list[np.ndarray]) – Detector data to integrate.

• poni_file (str) – Name of the [pyFAI PONI file] containing the detector properties pyFAI needs to perform azimuthal integration.

• npt (int) – Number of points in the output pattern.

• mask_file (str, optional) – File to use for masking the input data.

• integrate1d_kwargs (Optional[dict]) – Optional dictionary of keywords

Returns:

Azimuthal integration results as a dictionary of numpy arrays.

class RawDetectorDataMapProcessor(*, 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 return a map of raw detector data in a NeXus style NXroot object.

get_config(data)

Get instances of the map configuration object needed by this Processor.

Parameters:

data (list[PipelineData]) – Result of Reader.read where at least one item has the value ‘common.models.map.MapConfig’ for the ‘schema’ key.

Raises:

Exception – If a valid map config object cannot be constructed from data.

Returns:

Valid instance of the map configuration object with field values taken from data.

Return type:

MapConfig

get_nxroot(map_config, detector_name, detector_shape)

Get a map of the detector data collected by the scans in map_config. The data will be returned along with some relevant metadata in the form of a NeXus style NXroot structure.

Parameters:

• map_config (MapConfig) – Map configuration.

• detector_name (str) – Detector prefix.

• detector_shape (list) – Detector data shape for a single scan step.

Returns:

Map of the raw detector data.

Return type:

nexusformat.nexus.NXroot

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, detector_name, detector_shape)

Process configurations for a map and return the raw detector data data collected over the map.

Parameters:

• data (list[PipelineData]) – Input data.

• detector_name (str) – Detector prefix.

• detector_shape (list) – Detector data shape for a single scan step.

Returns:

Map of raw detector data.

Return type:

nexusformat.nexus.NXroot

class SetupNXdataProcessor(*, 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

Processor to set up and return an “empty” representation of a structured dataset. This representation will be an instance of a NeXus style NXdata object that has: A NeXus style NXfield entry for every coordinate/signal specified. nxaxes that are the NXfield entries for the coordinates and contain the values provided for each coordinate. NXfield entries of appropriate shape, but containing all zeros, for every signal. Attributes that define the axes, plus any additional attributes specified by the user.

This Processor is most useful as a “setup” step for constucting a representation of / container for a complete dataset that will be filled out in pieces later by UpdateNXdataProcessor.

add_data_point(data_point)

Add a data point to this dataset. 1. Validate data_point. 2. Append data_point to self.data_points. 3. Update signal NXfields in self.nxdata.

Parameters:

data_point (dict[str, object]) – Data point defining a point in the dataset’s coordinate space and the new signal values at that point.

get_index(data_point)

Return a tuple representing the array index of data_point in the coordinate space of the dataset.

Parameters:

data_point (dict[str, object]) – Data point defining a point in the dataset’s coordinate space.

Returns:

Multi-dimensional index of data_point in the dataset’s coordinate space.

Return type:

tuple

init_nxdata()

Initialize an empty NeXus style NXdata representing this dataset to self.nxdata; values for axes NXfield objects are filled out, values for signals’ NXfields are empty an can be filled out later.

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, nxname='data', coords=None, signals=None, attrs=None, data_points=None, extra_nxfields=None, duplicates='overwrite')

Return a NeXus style NXdata object that has the requisite axes and NXfield entries to represent a structured dataset with the properties provided. Properties may be provided either through the data argument (from an appropriate PipelineItem that immediately preceeds this one in a Pipeline), or through the coords, signals, attrs, and/or data_points arguments. If any of the latter are used, their values will completely override any values for these parameters found from data.

Parameters:

• data (list[PipelineData]) – Input data.

• nxname (str, optional) – Name for the returned NXdata object, defaults to ‘data’.

• coords (list[dict[str, object]], optional) – List of dictionaries defining the coordinates of the dataset. Each dictionary must have the keys ‘name’ and ‘values’, whose values are the name of the coordinate axis (a string) and all the unique values of that coordinate for the structured dataset (a list of numbers), respectively. A third item in the dictionary is optional, but highly recommended: ‘attrs’ may provide a dictionary of attributes to attach to the coordinate axis that assist in in interpreting the returned NXdata representation of the dataset. It is strongly recommended to provide the units of the values along an axis in the attrs dictionary.

• signals (list[dict[str, object]], optional) – List of dictionaries defining the signals of the dataset. Each dictionary must have the keys ‘name’ and ‘shape’, whose values are the name of the signal field (a string) and the shape of the signal’s value at each point in the dataset (a list of zero or more integers), respectively. A third item in the dictionary is optional, but highly recommended: ‘attrs’ may provide a dictionary of attributes to attach to the signal fieldthat assist in in interpreting the returned NXdata representation of the dataset. It is strongly recommended to provide the units of the signal’s values attrs dictionary.

• attrs (dict[str, object], optional) – An arbitrary dictionary of attributes to assign to the returned NXdata object.

• data_points (list[dict[str, object]], optional) – Data points to partially (or even entirely) fill out the “empty” signal NXfield’s before returning the NXdata object.

• extra_nxfields (list[dict[str, object]], optional) – List “extra” NXfields to include that can be described neither as a signal of the dataset, not a dedicated coordinate. This paramteter is good for including “alternate” values for one of the coordinate dimensions – the same coordinate axis expressed in different units, for instance. Each item in the list should be a dictionary of parameters for the nexusformat.nexus.NXfield constructor.

• duplicates (Literal['overwrite', 'block']) – Behavior to use if any new data points occur at the same point in the dataset’s coordinate space as an existing data point. Allowed values for duplicates are: ‘overwrite’ and ‘block’. Defaults to ‘overwrite’.

Returns:

Structured dataset as specified.

Return type:

nexusformat.nexus.NXdata

update_nxdata(data_point)

Update self.nxdata’s NXfield values.

Parameters:

data_point (dict[str, object]) – Data point defining a point in the dataset’s coordinate space and the new signal values at that point.

validate_data_point(data_point)

Return True if data_point occurs at a valid point in this structured dataset’s coordinate space, False otherwise. Also validate shapes of signal values and add NaN values for any missing signals.

Parameters:

data_point (dict[str, object]) – Data point defining a point in the dataset’s coordinate space and the new signal values at that point.

Returns:

Validity of data_point, message

Return type:

bool, str

class UnstructuredToStructuredProcessor(*, 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

Processor to reshape data in a NeXus style NXdata object from an “unstructured” to a “structured” representation.

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, nxpath=None)

Reshape the input data from an “unstructured” to a “structured” representation.

Parameters:

• data (list[PipelineData]) – Input data.

• nxname (str, optional) – Name for the returned NXdata object, defaults to ‘data’.

Returns:

Converted data.

Return type:

nexusformat.nexus.NXdata

class UpdateNXdataProcessor(*, 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

Processor to fill in part(s) of a NeXus style NXdata representing a structured dataset that’s already been written to a NeXus file.

This Processor is most useful as an “update” step for a NXdata object created by SetupNXdataProcessor, and is most easy to use in a Pipeline immediately after another PipelineItem designed specifically to return a value that can be used as input to this Processor.

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, nxfilename, nxdata_path, data_points=None, allow_approximate_coordinates=False)

Write new data points to the signal fields of an existing NXdata object representing a structued dataset in a NeXus file. Return the list of data points used to update the dataset.

Parameters:

• data (list[PipelineData]) – Data from the previous item in a Pipeline. May contain a list of data points that will extend the list of data points optionally provided with the data_points argument.

• nxfilename (str) – Name of the NeXus file containing the NXdata object to update.

• nxdata_path (str) – Path to the NXdata object to update in the file.

• data_points (Optional[list[dict[str, object]]]) – List of data points, each one a dictionary whose keys are the names of the coordinates and axes, and whose values are the values of each coordinate / signal at a single point in the dataset. Deafults to None.

• allow_approximate_coordinates (bool, optional) – Parameter to allow the nearest existing match for the new data points’ coordinates to be used if an exact match connot be found (sometimes this is due simply to differences in rounding convetions). Defaults to False.

Returns:

Complete list of data points used to update the dataset.

Return type:

list[dict[str, object]]

class UpdateNXvalueProcessor(*, 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

Processor to fill in part(s) of an object representing a structured dataset that’s already been written to a NeXus file.

This Processor is most useful as an “update” step for a NeXus style NXdata object created by SetupNXdataProcessor, and is most easy to use in a Pipeline immediately after another PipelineItem designed specifically to return a value that can be used as input to this Processor.

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, nxfilename, data_points=None)

Write new data values to an existing object representing an unstructured dataset in a NeXus file. Return the list of data points used to update the dataset.

Parameters:

• data (list[PipelineData]) – Data from the previous item in a Pipeline. May contain a list of data points that will extend the list of data points optionally provided with the data_points argument.

• nxfilename (str) – Name of the NeXus file containing the object to update.

• data_points (Optional[list[dict[str, object]]]) – List of data points, each one a dictionary whose keys are the names of the nxpath, the index of the data point in the dataset, and the data value.

Returns:

Complete list of data points used to update the dataset.

Return type:

list[dict[str, object]]

class XarrayToNexusProcessor(*, 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 the data in an xarray structure to a NeXus style NXdata object.

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)

Return the input data represented as a NeXus style NXdata object.

Parameters:

data (list[PipelineData]) – Input data.

Returns:

Data and metadata in data.

Return type:

nexusformat.nexus.NXdata

class XarrayToNumpyProcessor(*, 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 the data in an xarray.DataArray structure to an numpy.ndarray.

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)

Return just the signal values contained in data.

Parameters:

data (list[PipelineData]) – Input data.

Returns:

Data in data.

Return type:

numpy.ndarray

class ZarrToNexusProcessor(*, 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

Processor for converting Zarr data to NeXus file. format.

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, zarr_filename, nexus_filename)

Convert the signal values contained in the input data.

Parameters:

• data (list[PipelineData]) – Input data.

• zarr_filename (str) – Zarr input file name.

• nexus_filename (str) – NeXus output file name.

CHAP.common.reader module

Module for generic Readers used in multiple experiment-specific workflows.

class BinaryFileReader(*, 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: Reader

Reader for binary 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.

read()

Return a content of a given binary file.

Returns:

File content.

Return type:

binary

class ConfigReader(*, 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: Reader

Reader for YAML files that optionally implements and verifies it agaist its Pydantic configuration schema.

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.

read()

Return an optionally verified dictionary from the contents of a yaml file.

Returns:

File content.

Return type:

dict

class DetectorDataReader(*, 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: Reader

Reader for detector data files. Glob filenames allowed. Mask application and background correction available.

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.

read(filename, mask_file=None, mask_above=None, mask_below=None, mask_value=nan, data_scalar=None, background_file=None, background_scalar=None)

Reads detector data, applies masking, scaling, and background subtraction.

Parameters:

• filename (str) – Path to the primary data file.

• mask_file (str, optional) – Path to the mask file (optional).

• mask_above (float, optional) – Mask values above this threshold (optional).

• mask_below (float, optional) – Mask values below this threshold (optional).

• data_scalar (float, optional) – Scalar to multiply the data (optional).

• background_file (str, optional) – Path to the background file (optional).

• background_scalar (float, optional) – Scalar to multiply the background data (optional).

Returns:

Processed detector data.

Return type:

numpy.ndarray

class FabioImageReader(*, 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: Reader

Reader for images using the python fabio package.

Variables:

frame (int, optional) – Index of a specific frame to read from the file(s), defaults to None.

frame: Annotated[int, None, Interval(gt=None, ge=0, lt=None, le=None), 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.

read()

Return the data from the image file(s) provided.

Returns:

Image data as a numpy array (or list of numpy arrays, if a glob pattern matching more than one file was provided).

Return type:

numpy.ndarray | list[numpy.ndarray]

class H5Reader(*, 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: Reader

Reader for h5 files.

Variables:

• h5path (str, optional) – Path to a specific location in the h5 file to read data from, defaults to ‘/’.

• idx (list[int], optional) – Data slice to read from the object at the specified location in the h5 file.

h5path: 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

idx: Annotated[list[int], Len(min_length=1, max_length=3)] | 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.

read()

Return the data object stored at h5path in an h5-file.

Returns:

Object indicated by filename and h5path.

Return type:

Any

class LinkamReader(*, 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: Reader

Reader for loading Linkam load frame .txt files as an NXdata object.

Variables:

columns (list[str], optional) – Column names to read in, defaults to None (read in all columns)

columns: Annotated[list[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)]], Len(min_length=0, max_length=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.

classmethod parse_file(filename, logger)

Return start time, metadata, and data stored in the provided Linkam .txt file.

Returns:

Start time, metadata, and data stored in the input file.

Return type:

tuple(float, dict[str, str], dict[str, list[float]])

read()

Read specified columns from the given Linkam file.

Returns:

Linkam data.

Return type:

nexusformat.nexus.NXdata

class MapReader(*, 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: Reader

Reader for CHESS sample maps.

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.

read(filename=None, map_config=None, detector_names=None)

Take a map configuration dictionary and return a representation of the map as a NeXus style NXentry object. The NXentry’s default data group will contain the raw data collected over the course of the map.

Parameters:

• filename (str, optional) – Name of a file with the map configuration to read and pass onto the constructor of MapConfig.

• map_config (dict, optional) – Map configuration to be passed directly to the constructor of MapConfig.

• detector_names (list[str], optional) – Detector prefixes to include raw data for in the returned NXentry object.

Returns:

Data from the provided map configuration.

Return type:

nexusformat.nexus.NXentry

class NXdataReader(*, 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: Reader

Reader for constructing a NeXus style NXdata object from components.

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.

read(name, nxfield_params, signal_name, axes_names, attrs=None)

Return a basic NeXus style NXdata object constructed from components.

Parameters:

• name (str) – NXdata group name.

• nxfield_params (list[dict]) – List of sets of parameters for NXfieldReader specifying the NXfield objects belonging to the NXdata object.

• signal_name – Name of the signal for the NXdata (musts be one of the names of the NXfields indicated in nxfields).

• axes_names (str | list[str]) – Name or names of the coordinate axes NXfields associated with the signal (must be names of NXfields indicated in nxfields).

• attrs (dict, optional) – Additional configuration attributes.

Returns:

NXdata object.

Return type:

nexusformat.nexus.NXdata

class NXfieldReader(*, 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: Reader

Reader for a NeXus style NXfield with options to modify certain attributes.

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.

read(nxpath, nxname=None, update_attrs=None, slice_params=None)

Return a copy of the indicated NeXus style NXfield object from the file. Name and attributes of the returned copy may be modified with the nxname and update_attrs keyword arguments.

Parameters:

• nxpath (str) – Path in nxfile pointing to the NXfield to read.

• nxname (str, optional) – New name for the returned NXfield.

• update_attrs (dict, optional) – Optional dictonary used to add to / update the original NXfield’s attributes.

• slice_params (list[dict[str, int]], optional) – Parameters for returning just a slice of the full field data. Slice parameters are provided in a list dictionaries with integer values for any / all of the following keys: “start”, “end”, “step”. Default values used are: “start” - 0, “end” – None, “step” – 1. The order of the list must correspond to the order of the field’s axes.

Returns:

Copy of the indicated NXfield (with name and attributes optionally modified).

Return type:

nexusformat.nexus.NXfield

class NexusReader(*, 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: Reader

Reader for NeXus files.

Variables:

• nxpath (str, optional) – Path to a specific location in the NeXus file tree to read from, defaults to ‘/’.

• idx (int, optional) – Index of array to select, defaults to None

• mode (Literal['r', 'rw', 'r+', 'w', 'a'], optional) – File mode, defaults to ‘r’.

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

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

mode: Literal['r', 'rw', 'r+', 'w', 'a']

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

nxpath: 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

read()

Return the NeXus Style NXobject, object stored at nxpath in a NeXus file.

Raises:

nexusformat.nexus.NeXusError – If filename is not a NeXus file or nxpath is not in its tree.

Returns:

NeXus object indicated by filename and nxpath.

Return type:

nexusformat.nexus.NXobject

class PandasReader(*, 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: Reader

Reader for files that can be read in with pandas

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.

read(filename, method='read_csv', comment='#', kwargs=None)

Return a pandas.DataFrame read from the given file.

Parameters:

• filename (str) – Name of file to read from.

• method (str, optional) – Name of pandas method to use for reading from filename. Defaults to ‘read_csv’.

• comment (str, optional) – Character to identify comment lines in the input file, defaults to ‘#’.

• kwargs – Additional keyword arguments to supply to the pandas reader.

• kwargs – dict, optional.

Return type:

pandas.DataFrame

class SpecReader(*, 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: Reader

Reader for CHESS SPEC scans.

Variables:

• config (dict, optional) – SPEC configuration to be passed directly to the constructor of SpecConfig.

• detectors (dict | DetectorConfig, optional) – Detector configurations of the detectors to include raw data for in the returned NeXus NXroot object, defaults to None (only a valid input for EDD).

• filename (str, optional) – Name of file to read from.

config: dict | SpecConfig | None

detector_config: DetectorConfig | None

filename: str | 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.

read()

Take a SPEC configuration filename or dictionary and return the raw data as a NeXus style NXroot object.

Returns:

Data from the provided SPEC configuration.

Return type:

nexusformat.nexus.NXroot

validate_specreader_after()

Validate the SpecReader configuration.

Returns:

Validated configuration.

Return type:

SpecReader

class URLReader(*, 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: Reader

Reader for data available over HTTPS.

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.

read(url, headers=None, timeout=10)

Make an HTTPS request to the provided URL and return the results. Headers for the request are optional.

Parameters:

• url (str) – URL to read.

• headers (dict, optional) – Headers to attach to the request.

• timeout (int) – Timeout for the HTTPS request, defaults to 10.

Returns:

Content of the response.

Return type:

Any

class YAMLReader(*, 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: Reader

Reader for YAML 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.

read()

Return a dictionary from the contents of a yaml file.

Returns:

Contents of the file.

Return type:

dict

class ZarrReader(*, 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: Reader

Reader for Zarr stores.

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.

read(filename, path='/', idx=None, mode='r')

Return the Zarr object stored at path in a Zarr store.

Parameters:

• filename (str) – Path or URL to the Zarr store.

• path (str, optional) – Path to a specific location in the Zarr hierarchy to read from, defaults to ‘/’.

• idx (int, slice, tuple, optional) – Optional index or slice to apply to the returned object.

• mode (str, optional) – Store access mode, defaults to ‘r’. Common values: ‘r’, ‘r+’, ‘a’, ‘w’.

Raises:

KeyError – If path does not exist in the store.

Returns:

Zarr array or group indicated by filename and path.

Return type:

zarr.core.Array or zarr.hierarchy.Group

validate_model(model)

Validate the model configuration.

Returns:

Validated model.

Return type:

Any

CHAP.common.utils module

Some generic utility functions.

environments()

Detects the current Python environment (system, virtualenv, Conda, or pip) and collects package information. Returns a list of detected environments with installed packages.

Returns:

Information about the current Python environment.

Return type:

list[dict]

osinfo()

Helper function to provide the current osinfo.

Returns:

Information about the current OS.

Return type:

dict

CHAP.common.writer module

Module for generic Writers used in multiple experiment-specific workflows.

class ExtractArchiveWriter(*, 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 tar files from binary 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.

write(data)

Take a tar archive represented as bytes contained in data and write the extracted archive to files.

Parameters:

data (list[PipelineData]) – The data to write to archive.

class FileTreeWriter(*, 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: PipelineItem

Writer for a file tree in NeXus format.

Variables:

• force_overwrite (bool, optional) – Flag to allow data to be overwritten if it already exists, defaults to False. Note that the existence of files prior to executing the pipeline is not possible since the filename(s) of the data are unknown during pipeline validation.

• remove (bool, optional) – Flag to remove the dictionary from data, defaults to True.

force_overwrite: bool | 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.

remove: bool | None

write(data)

Write a NeXus format object contained in data to a directory tree stuctured like the NeXus tree.

Parameters:

data (list[PipelineData]) – The data to write to disk.

Raises:

RuntimeError – If filename already exists and force_overwrite is False.

class H5Writer(*, 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 HDF5 files from a NeXus style NXdata object.

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.

write(data)

Write the NeXus style NXdata object contained in data to hdf5 file.

Parameters:

data (list[PipelineData]) – The data to write to file.

Raises:

RuntimeError – If filename already exists and force_overwrite is False.

class ImageWriter(*, 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: PipelineItem

Writer for saving image files.

Variables:

• filename (str, optional) – Name of file to write to.

• force_overwrite (bool, optional) – Flag to allow data in filename to be overwritten if it already exists, defaults to False.

• remove (bool, optional) – Flag to remove the dictionary from data, defaults to False.

filename: str | None

force_overwrite: bool | 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.

remove: bool | None

write(data)

Write the image(s) contained in data to file.

Parameters:

data (list[PipelineData]) – The data to write to file.

Raises:

RuntimeError – If a file already exists and force_overwrite is False.

class MatplotlibAnimationWriter(*, 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 Matplotlib animations.

Variables:

fps (int, optional) – Movie frame rate (frames per second), defaults to 1.

fps: Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), 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.

write(data)

Write the matplotlib.animation.ArtistAnimation object contained in data to file.

Parameters:

data (list[PipelineData]) – The data to write to file.

class MatplotlibFigureWriter(*, 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 Matplotlib figures to image files.

Variables:

savefig_kw (dict, optional) – Keyword args to pass to matplotlib.figure.Figure.savefig.

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.

savefig_kw: dict | None

write(data)

Write the matplotlib.figure.Figure contained in data to file.

Parameters:

data (list[PipelineData]) – The data to write to file.

Raises:

RuntimeError – If filename already exists and force_overwrite is False.

class NexusValuesWriter(*, 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 updating values in an existing NeXus file.

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.

nxs_writer(nxroot, path, idx, data)

Write data to a specific NeXus file.

This method writes data to a specified dataset within a NeXus file at the given index (idx). If the dataset does not exist, an error is raised. The method ensures that the shape of data matches the shape of the target slice before writing.

Parameters:

• nxroot (nexusformat.nexus.NXroot) – NeXus Style NXroot object.

• path (str) – Path to the dataset inside the NeXus file.

• idx (tuple or int) – Index or slice where the data should be written.

• data (numpy.ndarray or compatible array-like object) – Data to be written to the specified slice in the dataset.

Raises:

ValueError – If the specified dataset does not exist or if the shape of data does not match the target slice.

write(data, filename, path_prefix='')

Write new values specified in data to the exising NeXus file filename.

Parameters:

• data (list[PipelineData]) – List of dictionaries with the following entries – ‘path’ identifying the location of the NeXus style NXfield object to which values will be written, ‘data’ identifying the data to be written, and ‘idx’ identifying the index / indicies of the NXfield to which the data will be written.

• filename (str) – Name of an existing NeXus file to update.

• path_prefix (str, optional) – Prefix to use for all paths in input data, defaults to ‘’.

class NexusWriter(*, 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 NeXus files from NeXus style NXobject objexts.

Variables:

nxpath (str, optional) – Path to a specific location in the NeXus file tree to write to (ignored if filename does not yet exist).

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.

nxpath: 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

write(data)

Write the NeXus style NXobject contained in data to file.

Parameters:

data (list[PipelineData]) – The data to write to file.

Raises:

RuntimeError – If filename already exists and force_overwrite is False.

class PyfaiResultsWriter(*, 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 results of one or more pyFAI integrations. Able to handle multiple output formats. Currently supported formats are: .npz, .nxs.

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.

write(data)

Save pyFAI integration results to a file. Format is determined automatically form the extension of filename.

Parameters:

data (list[PipelineData] | list[pyFAI.containers.IntegrateResult]) – The data to write to file.

write_npz(results, filename)

Save results to the .npz file, filename.

write_nxs(results, filename)

Save results to the .nxs file, filename.

class TXTWriter(*, 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 plain text files from string or tuples or lists of strings.

Variables:

append (bool, optional) – Flag to allow data in filename to be be appended, defaults to False.

append: bool | 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.

write(data)

Write a string or tuple or list of strings contained in data to file.

Parameters:

data (list[PipelineData]) – The data to write to file.

Raises:

• TypeError – If the object contained in data is not a str, tuple[str] or list[str].

• RuntimeError – If filename already exists and force_overwrite is False.

class YAMLWriter(*, 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 YAML files from dict-s.

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.

write(data)

Write the last matching dictionary contained in data to file (the schema mush match is a schema is provided).

Parameters:

data (list[PipelineData]) – The data to write to file.

Raises:

• TypeError – If the object contained in data is not a dict.

• RuntimeError – If filename already exists and force_overwrite is False.

class ZarrValuesWriter(*, 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 updating values in arrays of an existing Zarr file.

Variables:

path_prefix (str, optional) – Prefix to prepend to all “path” fields in data before writing. Defaults to “”.

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.

path_prefix: str | None

write(data)

Write values to specific paths and slices in an existing zarr file.

Parameters:

data (list[PipelineData]) – Data whose last item contains a list of dictionaries that each have three keys: “data”, “path”, “idx”.

zarr_writer(zarrfile, path, idx, data)

Write data to a specific dataset.

This method writes data to a specified dataset within a Zarr file at the given index (idx). If the dataset does not exist, an error is raised. The method ensures that the shape of data matches the shape of the target slice before writing.

Parameters:

• zarrfile (zarr.core.group.Group) – Path to the Zarr file.

• path (str) – Path to the dataset inside the Zarr file.

• idx (tuple or int) – Index or slice where the data should be written.

• data (numpy.ndarray or compatible array-like object) – Data to be written to the specified slice in the dataset.

Raises:

ValueError – If the specified dataset does not exist or if the shape of data does not match the target slice.

class ZarrWriter(*, 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 Zarr groups.

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.

write(data)

Write the last PipelineData item in data as text to a file.

Parameters:

data (list[PipelineData]) – Input data.

validate_model(model)

Validate the model configuration.

Returns:

Validated model.

Return type:

Any

write_filetree(data, outputdir='.', force_overwrite=False)

Write data to a file tree.

Parameters:

• data (nexusformat.nexus.NXobject) – The data to write to files

• outputdir – Output directory.

• force_overwrite (bool, optional) – Flag to allow data to be overwritten if it already exists, defaults to False.

write_matplotlibfigure(data, filename, savefig_kw, force_overwrite=False)

Write a Matplotlib figure to file.

Parameters:

• data (matplotlib.figure.Figure) – The figure to write to file

• filename (str) – File name.

• savefig_kw (dict, optional) – Keyword args to pass to matplotlib.figure.Figure.savefig.

• force_overwrite (bool, optional) – Flag to allow data to be overwritten if it already exists, defaults to False.

write_nexus(data, filename, force_overwrite=False)

Write a NeXus style NXobject object to file.

Parameters:

• data (nexusformat.nexus.NXobject) – The data to write to file

• filename (str) – File name.

• force_overwrite (bool, optional) – Flag to allow data to be overwritten if it already exists, defaults to False.

write_tif(data, filename, force_overwrite=False)

Write a tif image to file.

Parameters:

• data (numpy.ndarray) – The data to write to file

• filename (str) – File name.

• force_overwrite (bool, optional) – Flag to allow data to be overwritten if it already exists, defaults to False.

write_txt(data, filename, force_overwrite=False, append=False)

Write plain text to file.

Parameters:

• data (str | list[str]) – The data to write to file

• filename (str) – File name.

• force_overwrite (bool, optional) – Flag to allow data to be overwritten if it already exists, defaults to False.

• append (bool, optional) – Flag to allow data to be appended to the file if it already exists, defaults to False.

write_yaml(data, filename, force_overwrite=False)

Write data to a YAML file.

Parameters:

• data (dict | list) – The data to write to file

• filename (str) – File name.

• force_overwrite (bool, optional) – Flag to allow data to be overwritten if it already exists, defaults to False.