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
    • IndexSliceConfig
      • IndexSliceConfig.model_config
      • IndexSliceConfig.start
      • IndexSliceConfig.step
      • IndexSliceConfig.stop
    • 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.input_name
      • 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()
      • Detector.validate_shape()
    • 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()
      • MapConfig.validate_scalar_data_expressions()
    • 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

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).

• remove_constant_dims (bool, optional) – Flag to indicate that any independent_dimension`s in the map whose values are constant across the map should be exluded from the output `NXentry. Defaults to True.

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.

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

remove_constant_dims: bool | None

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 aggregate “unstructured” data into a single NeXus style NXdata object with a “structured” representation.

get_common_axes(signals)

Determine the common leading axis shared by all signals.

This method computes the longest common prefix of axis names across all signal definitions. Only axes that appear in the same order at the beginning of each signal’s axes list are included in the result.

This is used to identify the shared coordinate dimensions for a structured NeXus style NXdata <https://manual.nexusformat.org/classes/base_classes/NXdata.html#index-0>`__`NXdata object.

Parameters:

signals (list[dict]) – Validated signal definitions. Each signal must define an "axes" key containing an ordered list of axis names.

Returns:

Axis names that form the common leading axis for all signals. Returns an empty list if no common prefix exists.

Return type:

list[str]

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, fields, name='data', attrs=None)

Create and return a Nexus style NXdata object containing a single structured dataset composed from multiple unstructured input datasets.

This method validates the field configuration, validates and reshapes the input data, determines common axes across all signals, and constructs a NXdata group containing signal and axis fields.

Parameters:

• data (list[PipelineData]) – Input data objects containing unstructured datasets.

• fields (list[dict[str, Any]]) –

  Configuration describing how to structure the input data. This is a list of dictionaries. Each dictionary must contain the required keys:

  • "name": Name of the data item, which must correspond to the name field of an item in data.

  • "type": Either "signal" or "axis".

  • "axes": Required only for items where "type" is "signal". List of the names of the fields containing coordinate axes data for each dimension of the signal.

  Optional keys include: - "attrs": Dictionary of NeXus attributes to attach to.

• name (str, optional) – Name of the resulting NXdata group, defaults to “data”.

• attrs (dict[str, Any], optional) – Attributes to attach to the resulting NXdata group. The common axes determined during processing will be added to this dictionary under the "axes" key.

Returns:

A structured NXdata object containing all signals and axes defined by the configuration.

Return type:

nexusformat.nexus.NXdata

structure_signal_values(signals, axes, common_axes)

Reshape and populate structured signal arrays using common axes.

This method determines computes index mappings from raw axis values to their unique sorted representations, and inserts each signal’s unstructured data into its preallocated structured array.

Only the common axes are used for structuring; any trailing, signal-specific axes are assumed to have already been handled when allocating the structured signal arrays.

Parameters:

• signals (list[dict]) – Signal definitions with raw and preallocated structured data arrays.

• axes (list[dict]) – Axis definitions containing raw values and unique values.

• common_axes (list[str]) – Ordered list of the names of the dataset’s common axes.

Returns:

• Updated signal definitions with populated structured arrays.

• Unmodified axis definitions.

• Common axis names shared by all signals.

Return type:

tuple[list[dict], list[dict], list[str]]

validate_config_fields(fields)

Validate and normalize the field configuration.

This method separates the input field configuration into signal and axis definitions, performs basic validation, and ensures that all axes referenced by signals are defined as axis fields.

The returned signal and axis dictionaries are normalized into a consistent internal representation used by later processing stages.

Parameters:

fields (list[dict[str, Any]]) – Configuration describing how input data should be structured. Each item must define a "name" and "type" key, where "type" is either "signal" or "axis". Signal entries must additionally define an "axes" list.

Raises:

ValueError – If a signal references an axis that is not defined, or if a signal is defined before any axis exist.

Returns:

Validated signal and axis definitions.

Return type:

tuple[list[dict], list[dict]]

validate_data(data, signals, axes)

Validate and normalize input data for axes and signals.

This method retrieves raw input data for each axis and signal, propagates metadata attributes, computes unique axis values, and allocates structured arrays for signal data.

For each axis:

• Raw data is loaded.

• Attributes are merged (without overwriting user-specified ones).

• Unique axis values are computed.

For each signal:

• Raw data is loaded.

• Attributes are merged (without overwriting user-specified ones).

• A structured output array is allocated based on its axes.

• Total signal size is validated against the expected shape.

Parameters:

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

• signals (list[dict]) – Validated signal field definitions.

• axes (list[dict]) – Validated axis field definitions.

Raises:

ValueError – If a signal’s data size does not match the expected size derived from its axes.

Returns:

Updated signal and axis definitions with populated values and derived metadata.

Return type:

tuple[list[dict], list[dict]]

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:

• create_copy (bool, optional) – Return a copy of the selected data, resolving any and all linked NeXus objects, defaults to False..

• 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.

create_copy: bool | None

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)

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.

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.

• detector_config (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