CHAP.common package

Subpackages

• CHAP.common.models package
  • Submodules
  • CHAP.common.models.common module
    • BinarizeProcessorConfig
      • BinarizeProcessorConfig.method
      • BinarizeProcessorConfig.model_config
      • BinarizeProcessorConfig.num_bin
      • BinarizeProcessorConfig.nxpath
      • BinarizeProcessorConfig.remove_original_data
    • ImageProcessorConfig
      • ImageProcessorConfig.animation
      • ImageProcessorConfig.axis
      • ImageProcessorConfig.coord_range
      • ImageProcessorConfig.fileformat
      • ImageProcessorConfig.index_range
      • ImageProcessorConfig.model_config
      • ImageProcessorConfig.validate_index_range()
      • ImageProcessorConfig.validate_vrange()
      • ImageProcessorConfig.vrange
  • CHAP.common.models.integration module
    • Detector
      • Detector.azimuthal_integrator
      • Detector.mask_array
      • Detector.mask_file
      • Detector.model_config
      • Detector.poni_file
      • Detector.prefix
      • Detector.validate_mask_file()
      • Detector.validate_poni_file()
    • IntegrationConfig
      • IntegrationConfig.azimuthal_max
      • IntegrationConfig.azimuthal_min
      • IntegrationConfig.azimuthal_npt
      • IntegrationConfig.azimuthal_units
      • IntegrationConfig.detectors
      • IntegrationConfig.error_model
      • IntegrationConfig.get_azimuthal_adjustments()
      • IntegrationConfig.get_azimuthal_integrators()
      • IntegrationConfig.get_azimuthally_integrated_data()
      • IntegrationConfig.get_cake_integrated_data()
      • IntegrationConfig.get_integrated_data()
      • IntegrationConfig.get_multi_geometry_integrator()
      • IntegrationConfig.get_radially_integrated_data()
      • IntegrationConfig.integrated_data_coordinates
      • IntegrationConfig.integrated_data_dims
      • IntegrationConfig.integrated_data_shape
      • IntegrationConfig.integration_type
      • IntegrationConfig.model_config
      • IntegrationConfig.radial_max
      • IntegrationConfig.radial_min
      • IntegrationConfig.radial_npt
      • IntegrationConfig.radial_units
      • IntegrationConfig.sequence_index
      • IntegrationConfig.title
      • IntegrationConfig.tool_type
      • IntegrationConfig.validate_azimuthal_units()
      • IntegrationConfig.validate_for_map_config()
      • IntegrationConfig.validate_radial_units()
    • azimuthal_integrator()
    • get_azimuthal_adjustments()
    • get_azimuthal_integrators()
    • get_integrated_data_coordinates()
    • get_mask_array()
    • get_multi_geometry_integrator()
  • CHAP.common.models.map module
    • CorrectionsData
      • CorrectionsData.data_type
      • CorrectionsData.label
      • CorrectionsData.model_config
      • CorrectionsData.reserved_labels()
    • Detector
      • Detector.attrs
      • Detector.get_id()
      • Detector.id_
      • Detector.model_config
      • Detector.validate_attrs()
      • Detector.validate_id()
    • DetectorConfig
      • DetectorConfig.detectors
      • DetectorConfig.model_config
    • 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_experiment_type()
      • MapConfig.validate_mapconfig_before()
    • PointByPointScanData
      • PointByPointScanData.data_type
      • PointByPointScanData.get_value()
      • PointByPointScanData.label
      • PointByPointScanData.model_config
      • PointByPointScanData.name
      • PointByPointScanData.ndigits
      • PointByPointScanData.units
      • PointByPointScanData.validate_for_scalar_data()
      • PointByPointScanData.validate_for_spec_scans()
      • PointByPointScanData.validate_for_station()
      • PointByPointScanData.validate_label()
    • PostsampleIntensity
      • PostsampleIntensity.label
      • PostsampleIntensity.model_config
      • PostsampleIntensity.units
    • PresampleIntensity
      • PresampleIntensity.label
      • PresampleIntensity.model_config
      • PresampleIntensity.units
    • Sample
      • Sample.description
      • Sample.model_config
      • Sample.name
    • SpecConfig
      • SpecConfig.experiment_type
      • SpecConfig.model_config
      • SpecConfig.spec_scans
      • SpecConfig.station
      • SpecConfig.validate_experiment_type()
      • SpecConfig.validate_specconfig_before()
    • SpecScans
      • SpecScans.get_detector_data()
      • SpecScans.get_index()
      • SpecScans.get_scanparser()
      • SpecScans.model_config
      • SpecScans.par_file
      • SpecScans.scan_numbers
      • SpecScans.scanparsers
      • SpecScans.spec_file
      • SpecScans.validate_par_file()
      • SpecScans.validate_scan_numbers()
      • SpecScans.validate_spec_file()
    • get_available_scan_numbers()
    • get_detector_data()
    • get_detector_log_timestamps()
    • get_expression_value()
    • get_scanparser()
    • get_smb_par_value()
    • get_spec_counter_value()
    • get_spec_motor_value()
    • import_scanparser()
    • validate_data_source_for_map_config()
  • Module contents

Submodules

CHAP.common.processor module

File : processor.py Author : Valentin Kuznetsov <vkuznet AT gmail dot com> Description: Module for Processors used in multiple experiment-specific workflows.

class AsyncProcessor(mgr)

Bases: Processor

A Processor to process multiple sets of input data via asyncio module.

Variables:

mgr – 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 initialise 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 (iterable) – Input data documents to process.

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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

Bases: Processor

A Processor to binarize a 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 initialise 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)

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

Parameters:

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

• config (dict, optional) – Initialization parameters for an instance of CHAP.common.models.BinarizeProcessorConfig

Returns:

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

Return type:

Union[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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

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.array) – Input data.

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

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

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

• lam (float, optional) – The &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) – The 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:

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

Return type:

numpy.array, dict, Union[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 initialise 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 (list[PipelineData]) – Input data.

• x – Independent dimension (only used when running interactively or when filename is set).

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

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

• lam (float, optional) – The &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) – The 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:

The smoothed baseline and the configuration.

Return type:

numpy.array, 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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

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 initialise 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 – Input data.

Returns:

Processed data.

class ImageProcessor

Bases: Processor

A Processor to perform various visualization operations on images (slices) selected from a NeXus 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 initialise 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, save_figures=True)

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

Parameters:

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

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

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

Returns:

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

Return type:

Union[bytes, nexusformat.nexus.NXdata, numpy.ndarray]

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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

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

Returns a list of the distributed worker data on the root node.

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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

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 initialise 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 common.models.map.MapConfig.

• sub_pipeline (Pipeline, optional) – The sub-pipeline.

Returns:

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

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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

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 initialise 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) – The sub-pipeline.

Returns:

The 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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, pipeline_fields: dict = {'config': 'common.models.map.MapConfig', 'detector_config': 'common.models.map.DetectorConfig'}, config: MapConfig, detector_config: DetectorConfig, num_proc: Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), None] | None = 1)

Bases: Processor

A Processor that takes a map configuration and returns a NeXus NXentry object representing that map’s metadata and any scalar-valued raw data requested by the supplied map configuration.

Variables:

• config – Map configuration parameters to initialize an instance of common.models.map.MapConfig. Any values in ‘config’ supplant their corresponding values obtained from the pipeline data configuration.

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

• num_proc – Number of processors used to read map, defaults to 1.

config: MapConfig

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 initialise private attributes.

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

Args:

self: The BaseModel instance. context: The context.

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

pipeline_fields: dict

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

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

Parameters:

data – Input data.

Returns:

Processed data.

classmethod validate_num_proc(num_proc, info)

Validate the number of processors.

Variables:

num_proc – Number of processors used to read map, defaults to 1.

Returns:

Number of processors

Return type:

str

class NXdataToDataPointsProcessor(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

Bases: Processor

Transform a NeXus 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 initialise 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 pipeline data containing a NeXus NXdata object.

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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

Bases: Processor

A Processor to convert the default plottable data in a NeXus 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 initialise 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 object contained in data as an numpy.ndarray.

Parameters:

data (nexusformat.nexus.NXobject) – Input data.

Raises:

ValueError – If data has no default plottable data signal.

Returns:

The 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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

Bases: Processor

A Processor to convert the default plottable data in a NeXus 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 initialise 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 object contained in data as an xarray.DataArray.

Parameters:

data (nexusformat.nexus.NXobject) – Input data.

Raises:

ValueError – If metadata for xarray is absent from data

Returns:

The default plottable data signal.

Return type:

xarray.DataArray

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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

Bases: Processor

Processor for calling NormalizeNexusProcessor for (usually all) detector data in an NXroot resulting from 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 initialise 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 NXroot with additional fields containing normalized detector data.

Parameters:

• data (nexusformat.nexus.NXroot) – Input nexus structure containing all fields to be normalized an the field by which to normalize them.

• 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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

Bases: Processor

Processor for scaling one or more NXfields in the input nexus structure by the values of another NXfield in the same structure.

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 initialise 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 structure with additional fields containing the normalized data of each field in normalize_nxfields.

Parameters:

• data (nexusformat.nexus.NXgroup) – Input nexus structure containing all fields to be normalized an the field by which 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 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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

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 initialise 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 (object) – Input data.

Returns:

data

Return type:

object

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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

Bases: Processor

Processor to azimuthally integrate one or more frames of 2d detector data using the [pyFAI](https://pyfai.readthedocs.io/en/v2023.1/index.html) 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 initialise 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 (Union[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) – A 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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

Bases: Processor

A Processor to return a map of raw detector data in a NeXus 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:

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

Return type:

common.models.map.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 structure.

Parameters:

• map_config (common.models.map.MapConfig) – The map configuration.

• detector_name (str) – The detector prefix.

• detector_shape (list) – The shape of detector data for a single scan step.

Returns:

A 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 initialise 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 map configuration.

• detector_name (str) – The detector prefix.

• detector_shape (list) – The shape of detector data 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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

Bases: Processor

Processor to set up and return an “empty” NeXus representation of a structured dataset. This representation will be an instance of a NeXus NXdata object that has: A NeXus NXfield entry for every coordinate/signal specified. nxaxes that are the NeXus NXfield entries for the coordinates and contain the values provided for each coordinate. NeXus 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 NXfield`s 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.

Returns:

None

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 NXdata representing this dataset to self.nxdata; values for axes’ NXfield`s are filled out, values for signals’ `NXfield`s are empty an can be filled out later. Save the empty NeXus NXdata object to the NeXus file. Initialise `self.nxfile and self.nxdata_path with the NXFile object and actual nxpath used to save and make updates to the Nexus 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 initialise 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 NXdata object that has the requisite axes and NeXus 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]) – Data from the previous item in a Pipeline.

• nxname (str, optional) – Name for the returned NeXus 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 NeXus 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 NeXus 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 NeXus NXdata object.

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

• extra_nxfields (list[dict[str, object]], optional) – List “extra” NeXus 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:

A NeXus NXdata object that represents the 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.

Returns:

None

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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

Bases: Processor

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

convert_nxdata(nxdata)

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 initialise private attributes.

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

Args:

self: The BaseModel instance. context: The context.

process(data, nxpath=None)

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

Parameters:

data – Input data.

Returns:

Processed data.

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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

Bases: Processor

Processor to fill in part(s) of a NeXus 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 NeXus NXdata object created by common.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 initialise 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 NeXus 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 NeXus NXdata object to update.

• nxdata_path (str) – The path to the NeXus 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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

Bases: Processor

Processor to fill in part(s) of a NeXus 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 NXdata object created by common.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 initialise 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 NeXus 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 NeXus 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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

Bases: Processor

A Processor to convert the data in an xarray structure to a NeXus 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 initialise 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 data represented as a NeXus NXdata object.

Parameters:

data (Union[xarray.DataArray, xarray.Dataset]) – The input xarray structure.

Returns:

The 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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

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 initialise 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 (xarray.DataArray) – The input xarray.DataArray.

Returns:

The data in data.

Return type:

numpy.ndarray

CHAP.common.reader module

File : reader.py Author : Valentin Kuznetsov <vkuznet AT gmail dot com> Description: Module for 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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, filename: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)])

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 initialise 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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, filename: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)])

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

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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, filename: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)], frame: Annotated[int, None, Interval(gt=None, ge=0, lt=None, le=None), None] | None = None)

Bases: Reader

Reader for images using the python package.

Variables:

frame – 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 initialise 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:

Union[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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, filename: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)], h5path: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = '/', idx: Annotated[list[int], Len(min_length=1, max_length=3)] | None = None)

Bases: Reader

Reader for h5 files.

Variables:

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

• idx – 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)] | 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 initialise 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:

object

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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, filename: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)], 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)]], Len(min_length=0, max_length=None)] | None = None)

Bases: Reader

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

Variables:

columns – 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)]], 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 initialise 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:

Return type:

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

read()

Read specified columns from the given Linkam file.

Returns:

Linkam data represented in an NXdata object

Return type:

nexusformat.nexus.NXdata

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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, filename: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)])

Bases: Reader

Reader for constructing an 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 initialise 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 NXdata object constructed from components.

Parameters:

• name (str) – NXdata group name.

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

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

• axes_names (Union[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) – Dictionary of additional attributes for the NXdata.

Returns:

A new 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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, filename: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)])

Bases: Reader

Reader for an 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 initialise 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 NXfield 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:

A 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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, filename: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)], nxpath: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = '/', nxmemory: Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), None] | None = None)

Bases: Reader

Reader for NeXus files.

Variables:

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

• nxmemory – 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 initialise 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)] | None

read()

Return the NeXus 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 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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, filename: str | None = None, config: dict | SpecConfig | None = None, detector_config: DetectorConfig | None = None)

Bases: Reader

Reader for CHESS SPEC scans.

Variables:

• config – SPEC configuration to be passed directly to the constructor of CHAP.common.models.map.SpecConfig.

• detectors – 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 – 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 initialise 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 NXentry object.

Returns:

Data from the provided SPEC configuration.

Return type:

nexusformat.nexus.NXroot

validate_specreader_after()

Validate the SpecReader configuration.

Returns:

The validated configuration.

Return type:

PipelineItem

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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, filename: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)])

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

object

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', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, filename: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)])

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 initialise 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

validate_model(model)

CHAP.common.utils module

File : utils.py Author : Valentin Kuznetsov <vkuznet AT gmail dot com> Description: common set of 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.

osinfo()

Helper function to provide osinfo.

CHAP.common.writer module

File : writer.py Author : Valentin Kuznetsov <vkuznet AT gmail dot com> Description: Module for Writers used in multiple experiment-specific workflows.

class ExtractArchiveWriter(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, filename: str, force_overwrite: bool | None = False, remove: bool | None = False)

Bases: Writer

Writer for tar files from binary data.

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

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

model_post_init(context: Any, /) → None

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

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

Args:

self: The BaseModel instance. context: The context.

write(data)

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

Parameters:

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

Returns:

The achived data.

Return type:

bytes

class FileTreeWriter(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, force_overwrite: bool | None = False)

Bases: PipelineItem

Writer for a file tree in NeXus format.

Variables:

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

force_overwrite: bool | None

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

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

model_post_init(context: Any, /) → None

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

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

Args:

self: The BaseModel instance. context: The context.

write(data)

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

Parameters:

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

Raises:

RuntimeError – If filename already exists and force_overwrite is False.

Returns:

The data written to disk.

Return type:

Union[nexusformat.nexus.NXroot, nexusformat.nexus.NXentry]

class H5Writer(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, filename: str, force_overwrite: bool | None = False, remove: bool | None = False)

Bases: Writer

Writer for H5 files from an nexusformat.nexus.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 initialise private attributes.

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

Args:

self: The BaseModel instance. context: The context.

write(data)

Write the NeXus object contained in data to hdf5 file.

Parameters:

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

Raises:

RuntimeError – If filename already exists and force_overwrite is False.

Returns:

The data written to file.

Return type:

nexusformat.nexus.NXobject

class ImageWriter(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, filename: str | None = None, force_overwrite: bool | None = False, remove: bool | None = True)

Bases: PipelineItem

Writer for saving image files.

Variables:

• filename – Name of file to write to.

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

• remove – Flag to remove the dictionary from data, defaults to False.

filename: str | None

force_overwrite: bool | None

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

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

model_post_init(context: Any, /) → None

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

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

Args:

self: The BaseModel instance. context: The context.

remove: bool | None

write(data)

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

Parameters:

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

Raises:

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

Returns:

The data written to disk.

Return type:

list, dict, matplotlib.animation.FuncAnimation, numpy.ndarray

class MatplotlibAnimationWriter(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, filename: str, force_overwrite: bool | None = False, remove: bool | None = False, fps: Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), None] | None = 1)

Bases: Writer

Writer for saving matplotlib animations.

Variables:

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

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

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

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

model_post_init(context: Any, /) → None

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

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

Args:

self: The BaseModel instance. context: The context.

write(data)

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

Parameters:

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

Returns:

The original animation.

Return type:

matplotlib.animation.ArtistAnimation

class MatplotlibFigureWriter(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, filename: str, force_overwrite: bool | None = False, remove: bool | None = False, savefig_kw: dict | None = None)

Bases: Writer

Writer for saving matplotlib figures to image files.

Variables:

savefig_kw – Keyword args to pass to matplotlib.figure.Figure.savefig.

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

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

model_post_init(context: Any, /) → None

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

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

Args:

self: The BaseModel instance. context: The context.

savefig_kw: dict | None

write(data)

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

Parameters:

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

Raises:

RuntimeError – If filename already exists and force_overwrite is False.

Returns:

The original figure object.

Return type:

matplotlib.figure.Figure

class NexusWriter(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, filename: str, force_overwrite: bool | None = False, remove: bool | None = False, nxpath: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

Bases: Writer

Writer for NeXus files from NXobject-s.

Variables:

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

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

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

model_post_init(context: Any, /) → None

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

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

Args:

self: The BaseModel instance. context: The context.

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

write(data)

Write the NeXus object contained in data to file.

Parameters:

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

Raises:

RuntimeError – If filename already exists and force_overwrite is False.

Returns:

The data written to file.

Return type:

nexusformat.nexus.NXobject

class PyfaiResultsWriter(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, filename: str, force_overwrite: bool | None = False, remove: bool | None = False)

Bases: Writer

Writer for results of one or more pyFAI integrations. Able to handle multiple output formats. Currently supported formats are: .npz, .nxs.

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

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

model_post_init(context: Any, /) → None

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

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

Args:

self: The BaseModel instance. context: The context.

write(data)

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

Parameters:

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

write_npz(results, filename)

Save results to the .npz file, filename.

write_nxs(results, filename)

Save results to the .nxs file, filename.

class TXTWriter(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, filename: str, force_overwrite: bool | None = False, remove: bool | None = False, append: bool | None = False)

Bases: Writer

Writer for plain text files from string or tuples or lists of strings.

Variables:

append – Flag to allow data in filename to be be appended, defaults to False.

append: bool | None

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

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

model_post_init(context: Any, /) → None

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

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

Args:

self: The BaseModel instance. context: The context.

write(data)

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

Parameters:

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

Raises:

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

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

Returns:

The data written to file.

Return type:

str, tuple[str], list[str]

class YAMLWriter(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, filename: str, force_overwrite: bool | None = False, remove: bool | None = False)

Bases: Writer

Writer for YAML files from dict-s.

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

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

model_post_init(context: Any, /) → None

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

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

Args:

self: The BaseModel instance. context: The context.

write(data)

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

Parameters:

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

Raises:

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

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

Returns:

The data pipeline.

Return type:

list[PipelineData]

validate_model(model)

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

Write data to a file tree.

Parameters:

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

• outputdir – Output directory.

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

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

Write a Matplotlib figure to file.

Parameters:

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

• filename (str) – File name.

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

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

write_nexus(data, filename, force_overwrite=False)

Write a NeXus object to file.

Parameters:

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

• filename (str) – File name.

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

write_tif(data, filename, force_overwrite=False)

Write a tif image to file.

Parameters:

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

• filename (str) – File name.

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

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

Write plain text to file.

Parameters:

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

• filename (str) – File name.

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

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

write_yaml(data, filename, force_overwrite=False)

Write data to a YAML file.

Parameters:

• data (Union[dict, list]) – The data to write to file

• filename (str) – File name.

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

Module contents

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