analysis

analysis

Analysis

High-level orchestration of analysis tasks for a Project.

This class wires calculators and minimizers, exposes a compact interface for parameters, constraints and results, and coordinates computations across the project's sample models and experiments.

Typical usage:

• Display or filter parameters to fit.
• Select a calculator/minimizer implementation.
• Calculate patterns and run single or joint fits.

project: The parent Project object. aliases: A registry of human-friendly aliases for parameters. constraints: Symbolic constraints between parameters. calculator: Active calculator used for computations. fitter: Active fitter/minimizer driver.

Source code in src/easydiffraction/analysis/analysis.py

class Analysis:
    """High-level orchestration of analysis tasks for a Project.

    This class wires calculators and minimizers, exposes a compact
    interface for parameters, constraints and results, and coordinates
    computations across the project's sample models and experiments.

    Typical usage:

    - Display or filter parameters to fit.
    - Select a calculator/minimizer implementation.
    - Calculate patterns and run single or joint fits.

    Attributes:
    project: The parent Project object.
        aliases: A registry of human-friendly aliases for parameters.
        constraints: Symbolic constraints between parameters.
    calculator: Active calculator used for computations.
        fitter: Active fitter/minimizer driver.
    """

    _calculator = CalculatorFactory.create_calculator('cryspy')

    def __init__(self, project) -> None:
        """Create a new Analysis instance bound to a project.

        Args:
            project: The project that owns models and experiments.
        """
        self.project = project
        self.aliases = Aliases()
        self.constraints = Constraints()
        self.constraints_handler = ConstraintsHandler.get()
        self.calculator = Analysis._calculator  # Default calculator shared by project
        self._calculator_key: str = 'cryspy'  # Added to track the current calculator
        self._fit_mode: str = 'single'
        self.fitter = Fitter('lmfit (leastsq)')

    def _get_params_as_dataframe(
        self,
        params: List[Union[NumericDescriptor, Parameter]],
    ) -> pd.DataFrame:
        """Convert a list of parameters to a DataFrame.

        Args:
            params: List of DescriptorFloat or Parameter objects.

        Returns:
            A pandas DataFrame containing parameter information.
        """
        records = []
        for param in params:
            record = {}
            # TODO: Merge into one. Add field if attr exists
            # TODO: f'{param.value!r}' for StringDescriptor?
            if isinstance(param, (StringDescriptor, NumericDescriptor, Parameter)):
                record = {
                    ('fittable', 'left'): False,
                    ('datablock', 'left'): param._identity.datablock_entry_name,
                    ('category', 'left'): param._identity.category_code,
                    ('entry', 'left'): param._identity.category_entry_name or '',
                    ('parameter', 'left'): param.name,
                    ('value', 'right'): param.value,
                }
            if isinstance(param, (NumericDescriptor, Parameter)):
                record = record | {
                    ('units', 'left'): param.units,
                }
            if isinstance(param, Parameter):
                record = record | {
                    ('fittable', 'left'): True,
                    ('free', 'left'): param.free,
                    ('min', 'right'): param.fit_min,
                    ('max', 'right'): param.fit_max,
                    ('uncertainty', 'right'): param.uncertainty or '',
                }
            records.append(record)

        df = pd.DataFrame.from_records(records)
        df.columns = pd.MultiIndex.from_tuples(df.columns)
        return df

    def show_all_params(self) -> None:
        """Print a table with all parameters for sample models and
        experiments.
        """
        sample_models_params = self.project.sample_models.parameters
        experiments_params = self.project.experiments.parameters

        if not sample_models_params and not experiments_params:
            log.warning('No parameters found.')
            return

        tabler = TableRenderer.get()

        filtered_headers = [
            'datablock',
            'category',
            'entry',
            'parameter',
            'value',
            'fittable',
        ]

        console.paragraph('All parameters for all sample models (🧩 data blocks)')
        df = self._get_params_as_dataframe(sample_models_params)
        filtered_df = df[filtered_headers]
        tabler.render(filtered_df)

        console.paragraph('All parameters for all experiments (🔬 data blocks)')
        df = self._get_params_as_dataframe(experiments_params)
        filtered_df = df[filtered_headers]
        tabler.render(filtered_df)

    def show_fittable_params(self) -> None:
        """Print a table with parameters that can be included in
        fitting.
        """
        sample_models_params = self.project.sample_models.fittable_parameters
        experiments_params = self.project.experiments.fittable_parameters

        if not sample_models_params and not experiments_params:
            log.warning('No fittable parameters found.')
            return

        tabler = TableRenderer.get()

        filtered_headers = [
            'datablock',
            'category',
            'entry',
            'parameter',
            'value',
            'uncertainty',
            'units',
            'free',
        ]

        console.paragraph('Fittable parameters for all sample models (🧩 data blocks)')
        df = self._get_params_as_dataframe(sample_models_params)
        filtered_df = df[filtered_headers]
        tabler.render(filtered_df)

        console.paragraph('Fittable parameters for all experiments (🔬 data blocks)')
        df = self._get_params_as_dataframe(experiments_params)
        filtered_df = df[filtered_headers]
        tabler.render(filtered_df)

    def show_free_params(self) -> None:
        """Print a table with only currently-free (varying)
        parameters.
        """
        sample_models_params = self.project.sample_models.free_parameters
        experiments_params = self.project.experiments.free_parameters
        free_params = sample_models_params + experiments_params

        if not free_params:
            log.warning('No free parameters found.')
            return

        tabler = TableRenderer.get()

        filtered_headers = [
            'datablock',
            'category',
            'entry',
            'parameter',
            'value',
            'uncertainty',
            'min',
            'max',
            'units',
        ]

        console.paragraph(
            'Free parameters for both sample models (🧩 data blocks) '
            'and experiments (🔬 data blocks)'
        )
        df = self._get_params_as_dataframe(free_params)
        filtered_df = df[filtered_headers]
        tabler.render(filtered_df)

    def how_to_access_parameters(self) -> None:
        """Show Python access paths for all parameters.

        The output explains how to reference specific parameters in
        code.
        """
        sample_models_params = self.project.sample_models.parameters
        experiments_params = self.project.experiments.parameters
        all_params = {
            'sample_models': sample_models_params,
            'experiments': experiments_params,
        }

        if not all_params:
            log.warning('No parameters found.')
            return

        columns_headers = [
            'datablock',
            'category',
            'entry',
            'parameter',
            'How to Access in Python Code',
        ]

        columns_alignment = [
            'left',
            'left',
            'left',
            'left',
            'left',
        ]

        columns_data = []
        project_varname = self.project._varname
        for datablock_code, params in all_params.items():
            for param in params:
                if isinstance(param, (StringDescriptor, NumericDescriptor, Parameter)):
                    datablock_entry_name = param._identity.datablock_entry_name
                    category_code = param._identity.category_code
                    category_entry_name = param._identity.category_entry_name or ''
                    param_key = param.name
                    code_variable = (
                        f'{project_varname}.{datablock_code}'
                        f"['{datablock_entry_name}'].{category_code}"
                    )
                    if category_entry_name:
                        code_variable += f"['{category_entry_name}']"
                    code_variable += f'.{param_key}'
                    columns_data.append([
                        datablock_entry_name,
                        category_code,
                        category_entry_name,
                        param_key,
                        code_variable,
                    ])

        console.paragraph('How to access parameters')
        render_table(
            columns_headers=columns_headers,
            columns_alignment=columns_alignment,
            columns_data=columns_data,
        )

    def show_parameter_cif_uids(self) -> None:
        """Show CIF unique IDs for all parameters.

        The output explains which unique identifiers are used when
        creating CIF-based constraints.
        """
        sample_models_params = self.project.sample_models.parameters
        experiments_params = self.project.experiments.parameters
        all_params = {
            'sample_models': sample_models_params,
            'experiments': experiments_params,
        }

        if not all_params:
            log.warning('No parameters found.')
            return

        columns_headers = [
            'datablock',
            'category',
            'entry',
            'parameter',
            'Unique Identifier for CIF Constraints',
        ]

        columns_alignment = [
            'left',
            'left',
            'left',
            'left',
            'left',
        ]

        columns_data = []
        for _, params in all_params.items():
            for param in params:
                if isinstance(param, (StringDescriptor, NumericDescriptor, Parameter)):
                    datablock_entry_name = param._identity.datablock_entry_name
                    category_code = param._identity.category_code
                    category_entry_name = param._identity.category_entry_name or ''
                    param_key = param.name
                    cif_uid = param._cif_handler.uid
                    columns_data.append([
                        datablock_entry_name,
                        category_code,
                        category_entry_name,
                        param_key,
                        cif_uid,
                    ])

        console.paragraph('Show parameter CIF unique identifiers')
        render_table(
            columns_headers=columns_headers,
            columns_alignment=columns_alignment,
            columns_data=columns_data,
        )

    def show_current_calculator(self) -> None:
        """Print the name of the currently selected calculator
        engine.
        """
        console.paragraph('Current calculator')
        console.print(self.current_calculator)

    @staticmethod
    def show_supported_calculators() -> None:
        """Print a table of available calculator backends on this
        system.
        """
        CalculatorFactory.show_supported_calculators()

    @property
    def current_calculator(self) -> str:
        """The key/name of the active calculator backend."""
        return self._calculator_key

    @current_calculator.setter
    def current_calculator(self, calculator_name: str) -> None:
        """Switch to a different calculator backend.

        Args:
            calculator_name: Calculator key to use (e.g. 'cryspy').
        """
        calculator = CalculatorFactory.create_calculator(calculator_name)
        if calculator is None:
            return
        self.calculator = calculator
        self._calculator_key = calculator_name
        console.paragraph('Current calculator changed to')
        console.print(self.current_calculator)

    def show_current_minimizer(self) -> None:
        """Print the name of the currently selected minimizer."""
        console.paragraph('Current minimizer')
        console.print(self.current_minimizer)

    @staticmethod
    def show_available_minimizers() -> None:
        """Print a table of available minimizer drivers on this
        system.
        """
        MinimizerFactory.show_available_minimizers()

    @property
    def current_minimizer(self) -> Optional[str]:
        """The identifier of the active minimizer, if any."""
        return self.fitter.selection if self.fitter else None

    @current_minimizer.setter
    def current_minimizer(self, selection: str) -> None:
        """Switch to a different minimizer implementation.

        Args:
            selection: Minimizer selection string, e.g.
                'lmfit (leastsq)'.
        """
        self.fitter = Fitter(selection)
        console.paragraph('Current minimizer changed to')
        console.print(self.current_minimizer)

    @property
    def fit_mode(self) -> str:
        """Current fitting strategy: either 'single' or 'joint'."""
        return self._fit_mode

    @fit_mode.setter
    def fit_mode(self, strategy: str) -> None:
        """Set the fitting strategy.

        When set to 'joint', all experiments get default weights and
        are used together in a single optimization.

        Args:
                strategy: Either 'single' or 'joint'.

        Raises:
            ValueError: If an unsupported strategy value is
                provided.
        """
        if strategy not in ['single', 'joint']:
            raise ValueError("Fit mode must be either 'single' or 'joint'")
        self._fit_mode = strategy
        if strategy == 'joint' and not hasattr(self, 'joint_fit_experiments'):
            # Pre-populate all experiments with weight 0.5
            self.joint_fit_experiments = JointFitExperiments()
            for id in self.project.experiments.names:
                self.joint_fit_experiments.add_from_args(id=id, weight=0.5)
        console.paragraph('Current fit mode changed to')
        console.print(self._fit_mode)

    def show_available_fit_modes(self) -> None:
        """Print all supported fitting strategies and their
        descriptions.
        """
        strategies = [
            {
                'Strategy': 'single',
                'Description': 'Independent fitting of each experiment; no shared parameters',
            },
            {
                'Strategy': 'joint',
                'Description': 'Simultaneous fitting of all experiments; '
                'some parameters are shared',
            },
        ]

        columns_headers = ['Strategy', 'Description']
        columns_alignment = ['left', 'left']
        columns_data = []
        for item in strategies:
            strategy = item['Strategy']
            description = item['Description']
            columns_data.append([strategy, description])

        console.paragraph('Available fit modes')
        render_table(
            columns_headers=columns_headers,
            columns_alignment=columns_alignment,
            columns_data=columns_data,
        )

    def show_current_fit_mode(self) -> None:
        """Print the currently active fitting strategy."""
        console.paragraph('Current fit mode')
        console.print(self.fit_mode)

    def calculate_pattern(self, expt_name: str) -> None:
        """Calculate and store the diffraction pattern for an
        experiment.

        The pattern is stored in the target experiment's datastore.

        Args:
                expt_name: The identifier of the experiment to compute.
        """
        experiment = self.project.experiments[expt_name]
        sample_models = self.project.sample_models
        self.calculator.calculate_pattern(sample_models, experiment)

    def show_constraints(self) -> None:
        """Print a table of all user-defined symbolic constraints."""
        constraints_dict = dict(self.constraints)

        if not self.constraints._items:
            log.warning('No constraints defined.')
            return

        rows = []
        for constraint in constraints_dict.values():
            row = {
                'lhs_alias': constraint.lhs_alias.value,
                'rhs_expr': constraint.rhs_expr.value,
                'full expression': f'{constraint.lhs_alias.value} = {constraint.rhs_expr.value}',
            }
            rows.append(row)

        headers = ['lhs_alias', 'rhs_expr', 'full expression']
        alignments = ['left', 'left', 'left']
        rows = [[row[header] for header in headers] for row in rows]

        console.paragraph('User defined constraints')
        render_table(
            columns_headers=headers,
            columns_alignment=alignments,
            columns_data=rows,
        )

    def apply_constraints(self):
        """Apply the currently defined constraints to the active
        project.
        """
        if not self.constraints._items:
            log.warning('No constraints defined.')
            return

        self.constraints_handler.set_aliases(self.aliases)
        self.constraints_handler.set_constraints(self.constraints)
        self.constraints_handler.apply()

    def fit(self):
        """Execute fitting using the selected mode, calculator and
        minimizer.

        In 'single' mode, fits each experiment independently. In
        'joint' mode, performs a simultaneous fit across experiments
        with weights.
            Sets :attr:`fit_results` on success.
        """
        sample_models = self.project.sample_models
        if not sample_models:
            log.warning('No sample models found in the project. Cannot run fit.')
            return

        experiments = self.project.experiments
        if not experiments:
            log.warning('No experiments found in the project. Cannot run fit.')
            return

        calculator = self.calculator
        if not calculator:
            log.warning('No calculator is set. Cannot run fit.')
            return

        # Run the fitting process
        if self.fit_mode == 'joint':
            console.paragraph(
                f"Using all experiments 🔬 {experiments.names} for '{self.fit_mode}' fitting"
            )
            self.fitter.fit(
                sample_models,
                experiments,
                calculator,
                weights=self.joint_fit_experiments,
            )
        elif self.fit_mode == 'single':
            for expt_name in experiments.names:
                console.paragraph(
                    f"Using experiment 🔬 '{expt_name}' for '{self.fit_mode}' fitting"
                )
                experiment = experiments[expt_name]
                dummy_experiments = Experiments()  # TODO: Find a better name
                dummy_experiments.add(experiment)
                self.fitter.fit(sample_models, dummy_experiments, calculator)
        else:
            raise NotImplementedError(f'Fit mode {self.fit_mode} not implemented yet.')

        # After fitting, get the results
        self.fit_results = self.fitter.results

    def as_cif(self):
        """Serialize the analysis section to a CIF string.

        Returns:
            The analysis section represented as a CIF document string.
        """
        from easydiffraction.io.cif.serialize import analysis_to_cif

        return analysis_to_cif(self)

    def show_as_cif(self) -> None:
        """Render the analysis section as CIF in a formatted console
        view.
        """
        cif_text: str = self.as_cif()
        paragraph_title: str = 'Analysis 🧮 info as cif'
        console.paragraph(paragraph_title)
        render_cif(cif_text)

__init__(project)

Create a new Analysis instance bound to a project.

Parameters:

Name	Type	Description	Default
project		The project that owns models and experiments.	required

Source code in src/easydiffraction/analysis/analysis.py

def __init__(self, project) -> None:
    """Create a new Analysis instance bound to a project.

    Args:
        project: The project that owns models and experiments.
    """
    self.project = project
    self.aliases = Aliases()
    self.constraints = Constraints()
    self.constraints_handler = ConstraintsHandler.get()
    self.calculator = Analysis._calculator  # Default calculator shared by project
    self._calculator_key: str = 'cryspy'  # Added to track the current calculator
    self._fit_mode: str = 'single'
    self.fitter = Fitter('lmfit (leastsq)')

apply_constraints()

Apply the currently defined constraints to the active project.

Source code in src/easydiffraction/analysis/analysis.py

def apply_constraints(self):
    """Apply the currently defined constraints to the active
    project.
    """
    if not self.constraints._items:
        log.warning('No constraints defined.')
        return

    self.constraints_handler.set_aliases(self.aliases)
    self.constraints_handler.set_constraints(self.constraints)
    self.constraints_handler.apply()

as_cif()

Serialize the analysis section to a CIF string.

Returns:

Type	Description
	The analysis section represented as a CIF document string.

Source code in src/easydiffraction/analysis/analysis.py

def as_cif(self):
    """Serialize the analysis section to a CIF string.

    Returns:
        The analysis section represented as a CIF document string.
    """
    from easydiffraction.io.cif.serialize import analysis_to_cif

    return analysis_to_cif(self)

calculate_pattern(expt_name)

Calculate and store the diffraction pattern for an experiment.

The pattern is stored in the target experiment's datastore.

Parameters:

Name	Type	Description	Default
expt_name	str	The identifier of the experiment to compute.	required

Source code in src/easydiffraction/analysis/analysis.py

def calculate_pattern(self, expt_name: str) -> None:
    """Calculate and store the diffraction pattern for an
    experiment.

    The pattern is stored in the target experiment's datastore.

    Args:
            expt_name: The identifier of the experiment to compute.
    """
    experiment = self.project.experiments[expt_name]
    sample_models = self.project.sample_models
    self.calculator.calculate_pattern(sample_models, experiment)

current_calculator property writable

The key/name of the active calculator backend.

current_minimizer property writable

The identifier of the active minimizer, if any.

fit()

Execute fitting using the selected mode, calculator and minimizer.

In 'single' mode, fits each experiment independently. In 'joint' mode, performs a simultaneous fit across experiments with weights. Sets :attr:fit_results on success.

Source code in src/easydiffraction/analysis/analysis.py

def fit(self):
    """Execute fitting using the selected mode, calculator and
    minimizer.

    In 'single' mode, fits each experiment independently. In
    'joint' mode, performs a simultaneous fit across experiments
    with weights.
        Sets :attr:`fit_results` on success.
    """
    sample_models = self.project.sample_models
    if not sample_models:
        log.warning('No sample models found in the project. Cannot run fit.')
        return

    experiments = self.project.experiments
    if not experiments:
        log.warning('No experiments found in the project. Cannot run fit.')
        return

    calculator = self.calculator
    if not calculator:
        log.warning('No calculator is set. Cannot run fit.')
        return

    # Run the fitting process
    if self.fit_mode == 'joint':
        console.paragraph(
            f"Using all experiments 🔬 {experiments.names} for '{self.fit_mode}' fitting"
        )
        self.fitter.fit(
            sample_models,
            experiments,
            calculator,
            weights=self.joint_fit_experiments,
        )
    elif self.fit_mode == 'single':
        for expt_name in experiments.names:
            console.paragraph(
                f"Using experiment 🔬 '{expt_name}' for '{self.fit_mode}' fitting"
            )
            experiment = experiments[expt_name]
            dummy_experiments = Experiments()  # TODO: Find a better name
            dummy_experiments.add(experiment)
            self.fitter.fit(sample_models, dummy_experiments, calculator)
    else:
        raise NotImplementedError(f'Fit mode {self.fit_mode} not implemented yet.')

    # After fitting, get the results
    self.fit_results = self.fitter.results

fit_mode property writable

Current fitting strategy: either 'single' or 'joint'.

how_to_access_parameters()

Show Python access paths for all parameters.

The output explains how to reference specific parameters in code.

Source code in src/easydiffraction/analysis/analysis.py

def how_to_access_parameters(self) -> None:
    """Show Python access paths for all parameters.

    The output explains how to reference specific parameters in
    code.
    """
    sample_models_params = self.project.sample_models.parameters
    experiments_params = self.project.experiments.parameters
    all_params = {
        'sample_models': sample_models_params,
        'experiments': experiments_params,
    }

    if not all_params:
        log.warning('No parameters found.')
        return

    columns_headers = [
        'datablock',
        'category',
        'entry',
        'parameter',
        'How to Access in Python Code',
    ]

    columns_alignment = [
        'left',
        'left',
        'left',
        'left',
        'left',
    ]

    columns_data = []
    project_varname = self.project._varname
    for datablock_code, params in all_params.items():
        for param in params:
            if isinstance(param, (StringDescriptor, NumericDescriptor, Parameter)):
                datablock_entry_name = param._identity.datablock_entry_name
                category_code = param._identity.category_code
                category_entry_name = param._identity.category_entry_name or ''
                param_key = param.name
                code_variable = (
                    f'{project_varname}.{datablock_code}'
                    f"['{datablock_entry_name}'].{category_code}"
                )
                if category_entry_name:
                    code_variable += f"['{category_entry_name}']"
                code_variable += f'.{param_key}'
                columns_data.append([
                    datablock_entry_name,
                    category_code,
                    category_entry_name,
                    param_key,
                    code_variable,
                ])

    console.paragraph('How to access parameters')
    render_table(
        columns_headers=columns_headers,
        columns_alignment=columns_alignment,
        columns_data=columns_data,
    )

show_all_params()

Print a table with all parameters for sample models and experiments.

Source code in src/easydiffraction/analysis/analysis.py

def show_all_params(self) -> None:
    """Print a table with all parameters for sample models and
    experiments.
    """
    sample_models_params = self.project.sample_models.parameters
    experiments_params = self.project.experiments.parameters

    if not sample_models_params and not experiments_params:
        log.warning('No parameters found.')
        return

    tabler = TableRenderer.get()

    filtered_headers = [
        'datablock',
        'category',
        'entry',
        'parameter',
        'value',
        'fittable',
    ]

    console.paragraph('All parameters for all sample models (🧩 data blocks)')
    df = self._get_params_as_dataframe(sample_models_params)
    filtered_df = df[filtered_headers]
    tabler.render(filtered_df)

    console.paragraph('All parameters for all experiments (🔬 data blocks)')
    df = self._get_params_as_dataframe(experiments_params)
    filtered_df = df[filtered_headers]
    tabler.render(filtered_df)

show_as_cif()

Render the analysis section as CIF in a formatted console view.

Source code in src/easydiffraction/analysis/analysis.py

def show_as_cif(self) -> None:
    """Render the analysis section as CIF in a formatted console
    view.
    """
    cif_text: str = self.as_cif()
    paragraph_title: str = 'Analysis 🧮 info as cif'
    console.paragraph(paragraph_title)
    render_cif(cif_text)

show_available_fit_modes()

Print all supported fitting strategies and their descriptions.

Source code in src/easydiffraction/analysis/analysis.py

def show_available_fit_modes(self) -> None:
    """Print all supported fitting strategies and their
    descriptions.
    """
    strategies = [
        {
            'Strategy': 'single',
            'Description': 'Independent fitting of each experiment; no shared parameters',
        },
        {
            'Strategy': 'joint',
            'Description': 'Simultaneous fitting of all experiments; '
            'some parameters are shared',
        },
    ]

    columns_headers = ['Strategy', 'Description']
    columns_alignment = ['left', 'left']
    columns_data = []
    for item in strategies:
        strategy = item['Strategy']
        description = item['Description']
        columns_data.append([strategy, description])

    console.paragraph('Available fit modes')
    render_table(
        columns_headers=columns_headers,
        columns_alignment=columns_alignment,
        columns_data=columns_data,
    )

show_available_minimizers() staticmethod

Print a table of available minimizer drivers on this system.

Source code in src/easydiffraction/analysis/analysis.py

@staticmethod
def show_available_minimizers() -> None:
    """Print a table of available minimizer drivers on this
    system.
    """
    MinimizerFactory.show_available_minimizers()

show_constraints()

Print a table of all user-defined symbolic constraints.

Source code in src/easydiffraction/analysis/analysis.py

def show_constraints(self) -> None:
    """Print a table of all user-defined symbolic constraints."""
    constraints_dict = dict(self.constraints)

    if not self.constraints._items:
        log.warning('No constraints defined.')
        return

    rows = []
    for constraint in constraints_dict.values():
        row = {
            'lhs_alias': constraint.lhs_alias.value,
            'rhs_expr': constraint.rhs_expr.value,
            'full expression': f'{constraint.lhs_alias.value} = {constraint.rhs_expr.value}',
        }
        rows.append(row)

    headers = ['lhs_alias', 'rhs_expr', 'full expression']
    alignments = ['left', 'left', 'left']
    rows = [[row[header] for header in headers] for row in rows]

    console.paragraph('User defined constraints')
    render_table(
        columns_headers=headers,
        columns_alignment=alignments,
        columns_data=rows,
    )

show_current_calculator()

Print the name of the currently selected calculator engine.

Source code in src/easydiffraction/analysis/analysis.py

def show_current_calculator(self) -> None:
    """Print the name of the currently selected calculator
    engine.
    """
    console.paragraph('Current calculator')
    console.print(self.current_calculator)

show_current_fit_mode()

Print the currently active fitting strategy.

Source code in src/easydiffraction/analysis/analysis.py

def show_current_fit_mode(self) -> None:
    """Print the currently active fitting strategy."""
    console.paragraph('Current fit mode')
    console.print(self.fit_mode)

show_current_minimizer()

Print the name of the currently selected minimizer.

Source code in src/easydiffraction/analysis/analysis.py

def show_current_minimizer(self) -> None:
    """Print the name of the currently selected minimizer."""
    console.paragraph('Current minimizer')
    console.print(self.current_minimizer)

show_fittable_params()

Print a table with parameters that can be included in fitting.

Source code in src/easydiffraction/analysis/analysis.py

def show_fittable_params(self) -> None:
    """Print a table with parameters that can be included in
    fitting.
    """
    sample_models_params = self.project.sample_models.fittable_parameters
    experiments_params = self.project.experiments.fittable_parameters

    if not sample_models_params and not experiments_params:
        log.warning('No fittable parameters found.')
        return

    tabler = TableRenderer.get()

    filtered_headers = [
        'datablock',
        'category',
        'entry',
        'parameter',
        'value',
        'uncertainty',
        'units',
        'free',
    ]

    console.paragraph('Fittable parameters for all sample models (🧩 data blocks)')
    df = self._get_params_as_dataframe(sample_models_params)
    filtered_df = df[filtered_headers]
    tabler.render(filtered_df)

    console.paragraph('Fittable parameters for all experiments (🔬 data blocks)')
    df = self._get_params_as_dataframe(experiments_params)
    filtered_df = df[filtered_headers]
    tabler.render(filtered_df)

show_free_params()

Print a table with only currently-free (varying) parameters.

Source code in src/easydiffraction/analysis/analysis.py

def show_free_params(self) -> None:
    """Print a table with only currently-free (varying)
    parameters.
    """
    sample_models_params = self.project.sample_models.free_parameters
    experiments_params = self.project.experiments.free_parameters
    free_params = sample_models_params + experiments_params

    if not free_params:
        log.warning('No free parameters found.')
        return

    tabler = TableRenderer.get()

    filtered_headers = [
        'datablock',
        'category',
        'entry',
        'parameter',
        'value',
        'uncertainty',
        'min',
        'max',
        'units',
    ]

    console.paragraph(
        'Free parameters for both sample models (🧩 data blocks) '
        'and experiments (🔬 data blocks)'
    )
    df = self._get_params_as_dataframe(free_params)
    filtered_df = df[filtered_headers]
    tabler.render(filtered_df)

show_parameter_cif_uids()

Show CIF unique IDs for all parameters.

The output explains which unique identifiers are used when creating CIF-based constraints.

Source code in src/easydiffraction/analysis/analysis.py

def show_parameter_cif_uids(self) -> None:
    """Show CIF unique IDs for all parameters.

    The output explains which unique identifiers are used when
    creating CIF-based constraints.
    """
    sample_models_params = self.project.sample_models.parameters
    experiments_params = self.project.experiments.parameters
    all_params = {
        'sample_models': sample_models_params,
        'experiments': experiments_params,
    }

    if not all_params:
        log.warning('No parameters found.')
        return

    columns_headers = [
        'datablock',
        'category',
        'entry',
        'parameter',
        'Unique Identifier for CIF Constraints',
    ]

    columns_alignment = [
        'left',
        'left',
        'left',
        'left',
        'left',
    ]

    columns_data = []
    for _, params in all_params.items():
        for param in params:
            if isinstance(param, (StringDescriptor, NumericDescriptor, Parameter)):
                datablock_entry_name = param._identity.datablock_entry_name
                category_code = param._identity.category_code
                category_entry_name = param._identity.category_entry_name or ''
                param_key = param.name
                cif_uid = param._cif_handler.uid
                columns_data.append([
                    datablock_entry_name,
                    category_code,
                    category_entry_name,
                    param_key,
                    cif_uid,
                ])

    console.paragraph('Show parameter CIF unique identifiers')
    render_table(
        columns_headers=columns_headers,
        columns_alignment=columns_alignment,
        columns_data=columns_data,
    )

show_supported_calculators() staticmethod

Print a table of available calculator backends on this system.

Source code in src/easydiffraction/analysis/analysis.py

@staticmethod
def show_supported_calculators() -> None:
    """Print a table of available calculator backends on this
    system.
    """
    CalculatorFactory.show_supported_calculators()

calculation

Calculator

Invokes calculation engines for pattern generation.

Source code in src/easydiffraction/analysis/calculation.py

class Calculator:
    """Invokes calculation engines for pattern generation."""

    def __init__(self, engine: str = 'cryspy') -> None:
        """Initialize the diffraction calculator with a specified
        backend engine.

        Args:
            engine: Type of the calculation engine to use.
                    Supported types: 'crysfml', 'cryspy', 'pdffit'.
                    Default is 'cryspy'.
        """
        self.calculator_factory = CalculatorFactory()
        self._calculator = self.calculator_factory.create_calculator(engine)

    def set_calculator(self, engine: str) -> None:
        """Switch to a different calculator engine at runtime.

        Args:
            engine: New calculation engine type to use.
        """
        self._calculator = self.calculator_factory.create_calculator(engine)

    def calculate_structure_factors(
        self,
        sample_models: SampleModels,
        experiments: Experiments,
    ) -> Optional[List[Any]]:
        """Calculate HKL intensities (structure factors) for sample
        models and experiments.

        Args:
            sample_models: Collection of sample models.
            experiments: Collection of experiments.

        Returns:
            HKL intensities calculated by the backend calculator.
        """
        return self._calculator.calculate_structure_factors(sample_models, experiments)

    def calculate_pattern(
        self,
        sample_models: SampleModels,
        experiment: ExperimentBase,
    ) -> None:
        """Calculate diffraction pattern based on sample models and
        experiment. The calculated pattern is stored within the
        experiment's datastore.

        Args:
            sample_models: Collection of sample models.
            experiment: A single experiment object.
        """
        self._calculator.calculate_pattern(sample_models, experiment)

__init__(engine='cryspy')

Initialize the diffraction calculator with a specified backend engine.

Parameters:

Name	Type	Description	Default
engine	str	Type of the calculation engine to use. Supported types: 'crysfml', 'cryspy', 'pdffit'. Default is 'cryspy'.	'cryspy'

Source code in src/easydiffraction/analysis/calculation.py

def __init__(self, engine: str = 'cryspy') -> None:
    """Initialize the diffraction calculator with a specified
    backend engine.

    Args:
        engine: Type of the calculation engine to use.
                Supported types: 'crysfml', 'cryspy', 'pdffit'.
                Default is 'cryspy'.
    """
    self.calculator_factory = CalculatorFactory()
    self._calculator = self.calculator_factory.create_calculator(engine)

calculate_pattern(sample_models, experiment)

Calculate diffraction pattern based on sample models and experiment. The calculated pattern is stored within the experiment's datastore.

Parameters:

Name	Type	Description	Default
sample_models	SampleModels	Collection of sample models.	required
experiment	ExperimentBase	A single experiment object.	required

Source code in src/easydiffraction/analysis/calculation.py

def calculate_pattern(
    self,
    sample_models: SampleModels,
    experiment: ExperimentBase,
) -> None:
    """Calculate diffraction pattern based on sample models and
    experiment. The calculated pattern is stored within the
    experiment's datastore.

    Args:
        sample_models: Collection of sample models.
        experiment: A single experiment object.
    """
    self._calculator.calculate_pattern(sample_models, experiment)

calculate_structure_factors(sample_models, experiments)

Calculate HKL intensities (structure factors) for sample models and experiments.

Parameters:

Name	Type	Description	Default
sample_models	SampleModels	Collection of sample models.	required
experiments	Experiments	Collection of experiments.	required

Returns:

Type	Description
Optional[List[Any]]	HKL intensities calculated by the backend calculator.

Source code in src/easydiffraction/analysis/calculation.py

def calculate_structure_factors(
    self,
    sample_models: SampleModels,
    experiments: Experiments,
) -> Optional[List[Any]]:
    """Calculate HKL intensities (structure factors) for sample
    models and experiments.

    Args:
        sample_models: Collection of sample models.
        experiments: Collection of experiments.

    Returns:
        HKL intensities calculated by the backend calculator.
    """
    return self._calculator.calculate_structure_factors(sample_models, experiments)

set_calculator(engine)

Switch to a different calculator engine at runtime.

Parameters:

Name	Type	Description	Default
engine	str	New calculation engine type to use.	required

Source code in src/easydiffraction/analysis/calculation.py

def set_calculator(self, engine: str) -> None:
    """Switch to a different calculator engine at runtime.

    Args:
        engine: New calculation engine type to use.
    """
    self._calculator = self.calculator_factory.create_calculator(engine)

calculators

base

CalculatorBase

Bases: ABC

Base API for diffraction calculation engines.

Source code in src/easydiffraction/analysis/calculators/base.py

class CalculatorBase(ABC):
    """Base API for diffraction calculation engines."""

    @property
    @abstractmethod
    def name(self) -> str:
        pass

    @property
    @abstractmethod
    def engine_imported(self) -> bool:
        pass

    @abstractmethod
    def calculate_structure_factors(
        self,
        sample_model: SampleModelBase,
        experiment: ExperimentBase,
    ) -> None:
        """Calculate structure factors for a single sample model and
        experiment.
        """
        pass

    def calculate_pattern(
        self,
        sample_models: SampleModels,
        experiment: ExperimentBase,
        called_by_minimizer: bool = False,
    ) -> None:
        """Calculate the diffraction pattern for multiple sample models
        and a single experiment. The calculated pattern is stored within
        the experiment's datastore.

        Args:
            sample_models: Collection of sample models.
            experiment: The experiment object.
            called_by_minimizer: Whether the calculation is called by a
                minimizer.
        """
        x_data = experiment.datastore.x
        y_calc_zeros = np.zeros_like(x_data)

        valid_linked_phases = self._get_valid_linked_phases(sample_models, experiment)

        # Apply user constraints to all sample models
        constraints = ConstraintsHandler.get()
        constraints.apply()

        # Calculate contributions from valid linked sample models
        y_calc_scaled = y_calc_zeros
        for linked_phase in valid_linked_phases:
            sample_model_id = linked_phase._identity.category_entry_name
            sample_model_scale = linked_phase.scale.value
            sample_model = sample_models[sample_model_id]

            # Apply symmetry constraints
            sample_model.apply_symmetry_constraints()

            sample_model_y_calc = self._calculate_single_model_pattern(
                sample_model,
                experiment,
                called_by_minimizer=called_by_minimizer,
            )

            # if not sample_model_y_calc:
            #    return np.ndarray([])

            sample_model_y_calc_scaled = sample_model_scale * sample_model_y_calc
            y_calc_scaled += sample_model_y_calc_scaled

        # Calculate background contribution
        y_bkg = np.zeros_like(x_data)
        # TODO: Change to the following check in other places instead of
        #  old `hasattr` check, because `hasattr` triggers warnings?
        if 'background' in experiment._public_attrs():
            y_bkg = experiment.background.calculate(x_data)
        experiment.datastore.bkg = y_bkg

        # Calculate total pattern
        y_calc_total = y_calc_scaled + y_bkg
        experiment.datastore.calc = y_calc_total

    @abstractmethod
    def _calculate_single_model_pattern(
        self,
        sample_model: SampleModels,
        experiment: ExperimentBase,
        called_by_minimizer: bool,
    ) -> np.ndarray:
        """Calculate the diffraction pattern for a single sample model
        and experiment.

        Args:
            sample_model: The sample model object.
            experiment: The experiment object.
            called_by_minimizer: Whether the calculation is called by a
                minimizer.

        Returns:
            The calculated diffraction pattern as a NumPy array.
        """
        pass

    def _get_valid_linked_phases(
        self,
        sample_models: SampleModels,
        experiment: ExperimentBase,
    ) -> List[Any]:
        """Get valid linked phases from the experiment.

        Args:
            sample_models: Collection of sample models.
            experiment: The experiment object.

        Returns:
            A list of valid linked phases.
        """
        if not experiment.linked_phases:
            print('Warning: No linked phases found. Returning empty pattern.')
            return []

        valid_linked_phases = []
        for linked_phase in experiment.linked_phases:
            if linked_phase._identity.category_entry_name not in sample_models.names:
                print(
                    f"Warning: Linked phase '{linked_phase.id.value}' not "
                    f'found in Sample Models {sample_models.names}'
                )
                continue
            valid_linked_phases.append(linked_phase)

        if not valid_linked_phases:
            print(
                'Warning: None of the linked phases found in Sample '
                'Models. Returning empty pattern.'
            )

        return valid_linked_phases

calculate_pattern(sample_models, experiment, called_by_minimizer=False)

Calculate the diffraction pattern for multiple sample models and a single experiment. The calculated pattern is stored within the experiment's datastore.

Parameters:

Name	Type	Description	Default
sample_models	SampleModels	Collection of sample models.	required
experiment	ExperimentBase	The experiment object.	required
called_by_minimizer	bool	Whether the calculation is called by a minimizer.	False

Source code in src/easydiffraction/analysis/calculators/base.py

def calculate_pattern(
    self,
    sample_models: SampleModels,
    experiment: ExperimentBase,
    called_by_minimizer: bool = False,
) -> None:
    """Calculate the diffraction pattern for multiple sample models
    and a single experiment. The calculated pattern is stored within
    the experiment's datastore.

    Args:
        sample_models: Collection of sample models.
        experiment: The experiment object.
        called_by_minimizer: Whether the calculation is called by a
            minimizer.
    """
    x_data = experiment.datastore.x
    y_calc_zeros = np.zeros_like(x_data)

    valid_linked_phases = self._get_valid_linked_phases(sample_models, experiment)

    # Apply user constraints to all sample models
    constraints = ConstraintsHandler.get()
    constraints.apply()

    # Calculate contributions from valid linked sample models
    y_calc_scaled = y_calc_zeros
    for linked_phase in valid_linked_phases:
        sample_model_id = linked_phase._identity.category_entry_name
        sample_model_scale = linked_phase.scale.value
        sample_model = sample_models[sample_model_id]

        # Apply symmetry constraints
        sample_model.apply_symmetry_constraints()

        sample_model_y_calc = self._calculate_single_model_pattern(
            sample_model,
            experiment,
            called_by_minimizer=called_by_minimizer,
        )

        # if not sample_model_y_calc:
        #    return np.ndarray([])

        sample_model_y_calc_scaled = sample_model_scale * sample_model_y_calc
        y_calc_scaled += sample_model_y_calc_scaled

    # Calculate background contribution
    y_bkg = np.zeros_like(x_data)
    # TODO: Change to the following check in other places instead of
    #  old `hasattr` check, because `hasattr` triggers warnings?
    if 'background' in experiment._public_attrs():
        y_bkg = experiment.background.calculate(x_data)
    experiment.datastore.bkg = y_bkg

    # Calculate total pattern
    y_calc_total = y_calc_scaled + y_bkg
    experiment.datastore.calc = y_calc_total

calculate_structure_factors(sample_model, experiment) abstractmethod

Calculate structure factors for a single sample model and experiment.

Source code in src/easydiffraction/analysis/calculators/base.py

@abstractmethod
def calculate_structure_factors(
    self,
    sample_model: SampleModelBase,
    experiment: ExperimentBase,
) -> None:
    """Calculate structure factors for a single sample model and
    experiment.
    """
    pass

crysfml

CrysfmlCalculator

Bases: CalculatorBase

Wrapper for Crysfml library.

Source code in src/easydiffraction/analysis/calculators/crysfml.py

class CrysfmlCalculator(CalculatorBase):
    """Wrapper for Crysfml library."""

    engine_imported: bool = cfml_py_utilities is not None

    @property
    def name(self) -> str:
        return 'crysfml'

    def calculate_structure_factors(
        self,
        sample_models: SampleModels,
        experiments: Experiments,
    ) -> None:
        """Call Crysfml to calculate structure factors.

        Args:
            sample_models: The sample models to calculate structure
                factors for.
            experiments: The experiments associated with the sample
                models.
        """
        raise NotImplementedError('HKL calculation is not implemented for CrysfmlCalculator.')

    def _calculate_single_model_pattern(
        self,
        sample_model: SampleModels,
        experiment: ExperimentBase,
        called_by_minimizer: bool = False,
    ) -> Union[np.ndarray, List[float]]:
        """Calculates the diffraction pattern using Crysfml for the
        given sample model and experiment.

        Args:
            sample_model: The sample model to calculate the pattern for.
            experiment: The experiment associated with the sample model.
            called_by_minimizer: Whether the calculation is called by a
            minimizer.

        Returns:
            The calculated diffraction pattern as a NumPy array or a
                list of floats.
        """
        # Intentionally unused, required by public API/signature
        del called_by_minimizer

        crysfml_dict = self._crysfml_dict(sample_model, experiment)
        try:
            _, y = cfml_py_utilities.cw_powder_pattern_from_dict(crysfml_dict)
            y = self._adjust_pattern_length(y, len(experiment.datastore.x))
        except KeyError:
            print('[CrysfmlCalculator] Error: No calculated data')
            y = []
        return y

    def _adjust_pattern_length(
        self,
        pattern: List[float],
        target_length: int,
    ) -> List[float]:
        """Adjusts the length of the pattern to match the target length.

        Args:
            pattern: The pattern to adjust.
            target_length: The desired length of the pattern.

        Returns:
            The adjusted pattern.
        """
        # TODO: Check the origin of this discrepancy coming from
        #  PyCrysFML
        if len(pattern) > target_length:
            return pattern[:target_length]
        return pattern

    def _crysfml_dict(
        self,
        sample_model: SampleModels,
        experiment: ExperimentBase,
    ) -> Dict[str, Union[ExperimentBase, SampleModelBase]]:
        """Converts the sample model and experiment into a dictionary
        format for Crysfml.

        Args:
            sample_model: The sample model to convert.
            experiment: The experiment to convert.

        Returns:
            A dictionary representation of the sample model and
                experiment.
        """
        sample_model_dict = self._convert_sample_model_to_dict(sample_model)
        experiment_dict = self._convert_experiment_to_dict(experiment)
        return {
            'phases': [sample_model_dict],
            'experiments': [experiment_dict],
        }

    def _convert_sample_model_to_dict(
        self,
        sample_model: SampleModelBase,
    ) -> Dict[str, Any]:
        """Converts a sample model into a dictionary format.

        Args:
            sample_model: The sample model to convert.

        Returns:
            A dictionary representation of the sample model.
        """
        sample_model_dict = {
            sample_model.name: {
                '_space_group_name_H-M_alt': sample_model.space_group.name_h_m.value,
                '_cell_length_a': sample_model.cell.length_a.value,
                '_cell_length_b': sample_model.cell.length_b.value,
                '_cell_length_c': sample_model.cell.length_c.value,
                '_cell_angle_alpha': sample_model.cell.angle_alpha.value,
                '_cell_angle_beta': sample_model.cell.angle_beta.value,
                '_cell_angle_gamma': sample_model.cell.angle_gamma.value,
                '_atom_site': [],
            }
        }

        for atom in sample_model.atom_sites:
            atom_site = {
                '_label': atom.label.value,
                '_type_symbol': atom.type_symbol.value,
                '_fract_x': atom.fract_x.value,
                '_fract_y': atom.fract_y.value,
                '_fract_z': atom.fract_z.value,
                '_occupancy': atom.occupancy.value,
                '_adp_type': 'Biso',  # Assuming Biso for simplicity
                '_B_iso_or_equiv': atom.b_iso.value,
            }
            sample_model_dict[sample_model.name]['_atom_site'].append(atom_site)

        return sample_model_dict

    def _convert_experiment_to_dict(
        self,
        experiment: ExperimentBase,
    ) -> Dict[str, Any]:
        """Converts an experiment into a dictionary format.

        Args:
            experiment: The experiment to convert.

        Returns:
            A dictionary representation of the experiment.
        """
        expt_type = getattr(experiment, 'type', None)
        instrument = getattr(experiment, 'instrument', None)
        peak = getattr(experiment, 'peak', None)

        x_data = experiment.datastore.x
        twotheta_min = float(x_data.min())
        twotheta_max = float(x_data.max())

        # TODO: Process default values on the experiment creation
        #  instead of here
        exp_dict = {
            'NPD': {
                '_diffrn_radiation_probe': expt_type.radiation_probe.value
                if expt_type
                else 'neutron',
                '_diffrn_radiation_wavelength': instrument.setup_wavelength.value
                if instrument
                else 1.0,
                '_pd_instr_resolution_u': peak.broad_gauss_u.value if peak else 0.0,
                '_pd_instr_resolution_v': peak.broad_gauss_v.value if peak else 0.0,
                '_pd_instr_resolution_w': peak.broad_gauss_w.value if peak else 0.0,
                '_pd_instr_resolution_x': peak.broad_lorentz_x.value if peak else 0.0,
                '_pd_instr_resolution_y': peak.broad_lorentz_y.value if peak else 0.0,
                '_pd_meas_2theta_offset': instrument.calib_twotheta_offset.value
                if instrument
                else 0.0,
                '_pd_meas_2theta_range_min': twotheta_min,
                '_pd_meas_2theta_range_max': twotheta_max,
                '_pd_meas_2theta_range_inc': (twotheta_max - twotheta_min) / len(x_data),
            }
        }

        return exp_dict

calculate_structure_factors(sample_models, experiments)

Call Crysfml to calculate structure factors.

Parameters:

Name	Type	Description	Default
sample_models	SampleModels	The sample models to calculate structure factors for.	required
experiments	Experiments	The experiments associated with the sample models.	required

Source code in src/easydiffraction/analysis/calculators/crysfml.py

def calculate_structure_factors(
    self,
    sample_models: SampleModels,
    experiments: Experiments,
) -> None:
    """Call Crysfml to calculate structure factors.

    Args:
        sample_models: The sample models to calculate structure
            factors for.
        experiments: The experiments associated with the sample
            models.
    """
    raise NotImplementedError('HKL calculation is not implemented for CrysfmlCalculator.')

cryspy

CryspyCalculator

Bases: CalculatorBase

Cryspy-based diffraction calculator.

Converts EasyDiffraction models into Cryspy objects and computes patterns.

Source code in src/easydiffraction/analysis/calculators/cryspy.py

class CryspyCalculator(CalculatorBase):
    """Cryspy-based diffraction calculator.

    Converts EasyDiffraction models into Cryspy objects and computes
    patterns.
    """

    engine_imported: bool = cryspy is not None

    @property
    def name(self) -> str:
        return 'cryspy'

    def __init__(self) -> None:
        super().__init__()
        self._cryspy_dicts: Dict[str, Dict[str, Any]] = {}

    def calculate_structure_factors(
        self,
        sample_model: SampleModelBase,
        experiment: ExperimentBase,
    ) -> None:
        """Raises a NotImplementedError as HKL calculation is not
        implemented.

        Args:
            sample_model: The sample model to calculate structure
                factors for.
            experiment: The experiment associated with the sample
                models.
        """
        raise NotImplementedError('HKL calculation is not implemented for CryspyCalculator.')

    def _calculate_single_model_pattern(
        self,
        sample_model: SampleModelBase,
        experiment: ExperimentBase,
        called_by_minimizer: bool = False,
    ) -> Union[np.ndarray, List[float]]:
        """Calculates the diffraction pattern using Cryspy for the given
        sample model and experiment.

        We only recreate the cryspy_obj if this method is
         - NOT called by the minimizer, or
         - the cryspy_dict is NOT yet created.
        In other cases, we are modifying the existing cryspy_dict
        This allows significantly speeding up the calculation

        Args:
            sample_model: The sample model to calculate the pattern for.
            experiment: The experiment associated with the sample model.
            called_by_minimizer: Whether the calculation is called by a
            minimizer.

        Returns:
            The calculated diffraction pattern as a NumPy array or a
                list of floats.
        """
        combined_name = f'{sample_model.name}_{experiment.name}'

        if called_by_minimizer:
            if self._cryspy_dicts and combined_name in self._cryspy_dicts:
                cryspy_dict = self._recreate_cryspy_dict(sample_model, experiment)
            else:
                cryspy_obj = self._recreate_cryspy_obj(sample_model, experiment)
                cryspy_dict = cryspy_obj.get_dictionary()
        else:
            cryspy_obj = self._recreate_cryspy_obj(sample_model, experiment)
            cryspy_dict = cryspy_obj.get_dictionary()

        self._cryspy_dicts[combined_name] = copy.deepcopy(cryspy_dict)

        cryspy_in_out_dict: Dict[str, Any] = {}

        # Calculate the pattern using Cryspy
        # TODO: Redirect stderr to suppress Cryspy warnings.
        #  This is a temporary solution to avoid cluttering the output.
        #  E.g. cryspy/A_functions_base/powder_diffraction_tof.py:106:
        #  RuntimeWarning: overflow encountered in exp
        #  Remove this when Cryspy is updated to handle warnings better.
        with contextlib.redirect_stderr(io.StringIO()):
            rhochi_calc_chi_sq_by_dictionary(
                cryspy_dict,
                dict_in_out=cryspy_in_out_dict,
                flag_use_precalculated_data=False,
                flag_calc_analytical_derivatives=False,
            )

        prefixes = {
            BeamModeEnum.CONSTANT_WAVELENGTH: 'pd',
            BeamModeEnum.TIME_OF_FLIGHT: 'tof',
        }
        beam_mode = experiment.type.beam_mode.value
        if beam_mode in prefixes:
            cryspy_block_name = f'{prefixes[beam_mode]}_{experiment.name}'
        else:
            print(f'[CryspyCalculator] Error: Unknown beam mode {experiment.type.beam_mode.value}')
            return []

        try:
            signal_plus = cryspy_in_out_dict[cryspy_block_name]['signal_plus']
            signal_minus = cryspy_in_out_dict[cryspy_block_name]['signal_minus']
            y_calc = signal_plus + signal_minus
        except KeyError:
            print(f'[CryspyCalculator] Error: No calculated data for {cryspy_block_name}')
            return []

        return y_calc

    def _recreate_cryspy_dict(
        self,
        sample_model: SampleModelBase,
        experiment: ExperimentBase,
    ) -> Dict[str, Any]:
        """Recreates the Cryspy dictionary for the given sample model
        and experiment.

        Args:
            sample_model: The sample model to update.
            experiment: The experiment to update.

        Returns:
            The updated Cryspy dictionary.
        """
        combined_name = f'{sample_model.name}_{experiment.name}'
        cryspy_dict = copy.deepcopy(self._cryspy_dicts[combined_name])

        cryspy_model_id = f'crystal_{sample_model.name}'
        cryspy_model_dict = cryspy_dict[cryspy_model_id]

        # Update sample model parameters

        # Cell
        cryspy_cell = cryspy_model_dict['unit_cell_parameters']
        cryspy_cell[0] = sample_model.cell.length_a.value
        cryspy_cell[1] = sample_model.cell.length_b.value
        cryspy_cell[2] = sample_model.cell.length_c.value
        cryspy_cell[3] = np.deg2rad(sample_model.cell.angle_alpha.value)
        cryspy_cell[4] = np.deg2rad(sample_model.cell.angle_beta.value)
        cryspy_cell[5] = np.deg2rad(sample_model.cell.angle_gamma.value)

        # Atomic coordinates
        cryspy_xyz = cryspy_model_dict['atom_fract_xyz']
        for idx, atom_site in enumerate(sample_model.atom_sites):
            cryspy_xyz[0][idx] = atom_site.fract_x.value
            cryspy_xyz[1][idx] = atom_site.fract_y.value
            cryspy_xyz[2][idx] = atom_site.fract_z.value

        # Atomic occupancies
        cryspy_occ = cryspy_model_dict['atom_occupancy']
        for idx, atom_site in enumerate(sample_model.atom_sites):
            cryspy_occ[idx] = atom_site.occupancy.value

        # Atomic ADPs - Biso only for now
        cryspy_biso = cryspy_model_dict['atom_b_iso']
        for idx, atom_site in enumerate(sample_model.atom_sites):
            cryspy_biso[idx] = atom_site.b_iso.value

        # Update experiment parameters

        if experiment.type.beam_mode.value == BeamModeEnum.CONSTANT_WAVELENGTH:
            cryspy_expt_name = f'pd_{experiment.name}'
            cryspy_expt_dict = cryspy_dict[cryspy_expt_name]

            # Instrument
            cryspy_expt_dict['offset_ttheta'][0] = np.deg2rad(
                experiment.instrument.calib_twotheta_offset.value
            )
            cryspy_expt_dict['wavelength'][0] = experiment.instrument.setup_wavelength.value

            # Peak
            cryspy_resolution = cryspy_expt_dict['resolution_parameters']
            cryspy_resolution[0] = experiment.peak.broad_gauss_u.value
            cryspy_resolution[1] = experiment.peak.broad_gauss_v.value
            cryspy_resolution[2] = experiment.peak.broad_gauss_w.value
            cryspy_resolution[3] = experiment.peak.broad_lorentz_x.value
            cryspy_resolution[4] = experiment.peak.broad_lorentz_y.value

        elif experiment.type.beam_mode.value == BeamModeEnum.TIME_OF_FLIGHT:
            cryspy_expt_name = f'tof_{experiment.name}'
            cryspy_expt_dict = cryspy_dict[cryspy_expt_name]

            # Instrument
            cryspy_expt_dict['zero'][0] = experiment.instrument.calib_d_to_tof_offset.value
            cryspy_expt_dict['dtt1'][0] = experiment.instrument.calib_d_to_tof_linear.value
            cryspy_expt_dict['dtt2'][0] = experiment.instrument.calib_d_to_tof_quad.value
            cryspy_expt_dict['ttheta_bank'] = np.deg2rad(
                experiment.instrument.setup_twotheta_bank.value
            )

            # Peak
            cryspy_sigma = cryspy_expt_dict['profile_sigmas']
            cryspy_sigma[0] = experiment.peak.broad_gauss_sigma_0.value
            cryspy_sigma[1] = experiment.peak.broad_gauss_sigma_1.value
            cryspy_sigma[2] = experiment.peak.broad_gauss_sigma_2.value

            cryspy_beta = cryspy_expt_dict['profile_betas']
            cryspy_beta[0] = experiment.peak.broad_mix_beta_0.value
            cryspy_beta[1] = experiment.peak.broad_mix_beta_1.value

            cryspy_alpha = cryspy_expt_dict['profile_alphas']
            cryspy_alpha[0] = experiment.peak.asym_alpha_0.value
            cryspy_alpha[1] = experiment.peak.asym_alpha_1.value

        return cryspy_dict

    def _recreate_cryspy_obj(
        self,
        sample_model: SampleModelBase,
        experiment: ExperimentBase,
    ) -> Any:
        """Recreates the Cryspy object for the given sample model and
        experiment.

        Args:
            sample_model: The sample model to recreate.
            experiment: The experiment to recreate.

        Returns:
            The recreated Cryspy object.
        """
        cryspy_obj = str_to_globaln('')

        cryspy_sample_model_cif = self._convert_sample_model_to_cryspy_cif(sample_model)
        cryspy_sample_model_obj = str_to_globaln(cryspy_sample_model_cif)
        cryspy_obj.add_items(cryspy_sample_model_obj.items)

        # Add single experiment to cryspy_obj
        cryspy_experiment_cif = self._convert_experiment_to_cryspy_cif(
            experiment,
            linked_phase=sample_model,
        )

        cryspy_experiment_obj = str_to_globaln(cryspy_experiment_cif)
        cryspy_obj.add_items(cryspy_experiment_obj.items)

        return cryspy_obj

    def _convert_sample_model_to_cryspy_cif(
        self,
        sample_model: SampleModelBase,
    ) -> str:
        """Converts a sample model to a Cryspy CIF string.

        Args:
            sample_model: The sample model to convert.

        Returns:
            The Cryspy CIF string representation of the sample model.
        """
        return sample_model.as_cif

    def _convert_experiment_to_cryspy_cif(
        self,
        experiment: ExperimentBase,
        linked_phase: Any,
    ) -> str:
        """Converts an experiment to a Cryspy CIF string.

        Args:
            experiment: The experiment to convert.
            linked_phase: The linked phase associated with the
                experiment.

        Returns:
            The Cryspy CIF string representation of the experiment.
        """
        expt_type = getattr(experiment, 'type', None)
        instrument = getattr(experiment, 'instrument', None)
        peak = getattr(experiment, 'peak', None)

        cif_lines = [f'data_{experiment.name}']

        if expt_type is not None:
            cif_lines.append('')
            radiation_probe = expt_type.radiation_probe.value
            radiation_probe = radiation_probe.replace('neutron', 'neutrons')
            radiation_probe = radiation_probe.replace('xray', 'X-rays')
            cif_lines.append(f'_setup_radiation {radiation_probe}')

        if instrument:
            # Restrict to only attributes relevant for the beam mode to
            # avoid probing non-existent guarded attributes (which
            # triggers diagnostics).
            if expt_type.beam_mode.value == BeamModeEnum.CONSTANT_WAVELENGTH:
                instrument_mapping = {
                    'setup_wavelength': '_setup_wavelength',
                    'calib_twotheta_offset': '_setup_offset_2theta',
                }
            elif expt_type.beam_mode.value == BeamModeEnum.TIME_OF_FLIGHT:
                instrument_mapping = {
                    'setup_twotheta_bank': '_tof_parameters_2theta_bank',
                    'calib_d_to_tof_offset': '_tof_parameters_Zero',
                    'calib_d_to_tof_linear': '_tof_parameters_Dtt1',
                    'calib_d_to_tof_quad': '_tof_parameters_dtt2',
                }
            cif_lines.append('')
            for local_attr_name, engine_key_name in instrument_mapping.items():
                # attr_obj = instrument.__dict__.get(local_attr_name)
                attr_obj = getattr(instrument, local_attr_name)
                if attr_obj is not None:
                    cif_lines.append(f'{engine_key_name} {attr_obj.value}')

        if peak:
            if expt_type.beam_mode.value == BeamModeEnum.CONSTANT_WAVELENGTH:
                peak_mapping = {
                    'broad_gauss_u': '_pd_instr_resolution_U',
                    'broad_gauss_v': '_pd_instr_resolution_V',
                    'broad_gauss_w': '_pd_instr_resolution_W',
                    'broad_lorentz_x': '_pd_instr_resolution_X',
                    'broad_lorentz_y': '_pd_instr_resolution_Y',
                }
            elif expt_type.beam_mode.value == BeamModeEnum.TIME_OF_FLIGHT:
                peak_mapping = {
                    'broad_gauss_sigma_0': '_tof_profile_sigma0',
                    'broad_gauss_sigma_1': '_tof_profile_sigma1',
                    'broad_gauss_sigma_2': '_tof_profile_sigma2',
                    'broad_mix_beta_0': '_tof_profile_beta0',
                    'broad_mix_beta_1': '_tof_profile_beta1',
                    'asym_alpha_0': '_tof_profile_alpha0',
                    'asym_alpha_1': '_tof_profile_alpha1',
                }
                cif_lines.append('_tof_profile_peak_shape Gauss')
            cif_lines.append('')
            for local_attr_name, engine_key_name in peak_mapping.items():
                # attr_obj = peak.__dict__.get(local_attr_name)
                attr_obj = getattr(peak, local_attr_name)
                if attr_obj is not None:
                    cif_lines.append(f'{engine_key_name} {attr_obj.value}')

        x_data = experiment.datastore.x
        twotheta_min = float(x_data.min())
        twotheta_max = float(x_data.max())
        cif_lines.append('')
        if expt_type.beam_mode.value == BeamModeEnum.CONSTANT_WAVELENGTH:
            cif_lines.append(f'_range_2theta_min {twotheta_min}')
            cif_lines.append(f'_range_2theta_max {twotheta_max}')
        elif expt_type.beam_mode.value == BeamModeEnum.TIME_OF_FLIGHT:
            cif_lines.append(f'_range_time_min {twotheta_min}')
            cif_lines.append(f'_range_time_max {twotheta_max}')

        cif_lines.append('')
        cif_lines.append('loop_')
        cif_lines.append('_phase_label')
        cif_lines.append('_phase_scale')
        cif_lines.append(f'{linked_phase.name} 1.0')

        if expt_type.beam_mode.value == BeamModeEnum.CONSTANT_WAVELENGTH:
            cif_lines.append('')
            cif_lines.append('loop_')
            cif_lines.append('_pd_background_2theta')
            cif_lines.append('_pd_background_intensity')
            cif_lines.append(f'{twotheta_min} 0.0')
            cif_lines.append(f'{twotheta_max} 0.0')
        elif expt_type.beam_mode.value == BeamModeEnum.TIME_OF_FLIGHT:
            cif_lines.append('')
            cif_lines.append('loop_')
            cif_lines.append('_tof_backgroundpoint_time')
            cif_lines.append('_tof_backgroundpoint_intensity')
            cif_lines.append(f'{twotheta_min} 0.0')
            cif_lines.append(f'{twotheta_max} 0.0')

        if expt_type.beam_mode.value == BeamModeEnum.CONSTANT_WAVELENGTH:
            cif_lines.append('')
            cif_lines.append('loop_')
            cif_lines.append('_pd_meas_2theta')
            cif_lines.append('_pd_meas_intensity')
            cif_lines.append('_pd_meas_intensity_sigma')
        elif expt_type.beam_mode.value == BeamModeEnum.TIME_OF_FLIGHT:
            cif_lines.append('')
            cif_lines.append('loop_')
            cif_lines.append('_tof_meas_time')
            cif_lines.append('_tof_meas_intensity')
            cif_lines.append('_tof_meas_intensity_sigma')

        y_data = experiment.datastore.meas
        sy_data = experiment.datastore.meas_su
        for x_val, y_val, sy_val in zip(x_data, y_data, sy_data, strict=True):
            cif_lines.append(f'  {x_val:.5f}   {y_val:.5f}   {sy_val:.5f}')

        cryspy_experiment_cif = '\n'.join(cif_lines)

        return cryspy_experiment_cif

calculate_structure_factors(sample_model, experiment)

Raises a NotImplementedError as HKL calculation is not implemented.

Parameters:

Name	Type	Description	Default
sample_model	SampleModelBase	The sample model to calculate structure factors for.	required
experiment	ExperimentBase	The experiment associated with the sample models.	required

Source code in src/easydiffraction/analysis/calculators/cryspy.py

def calculate_structure_factors(
    self,
    sample_model: SampleModelBase,
    experiment: ExperimentBase,
) -> None:
    """Raises a NotImplementedError as HKL calculation is not
    implemented.

    Args:
        sample_model: The sample model to calculate structure
            factors for.
        experiment: The experiment associated with the sample
            models.
    """
    raise NotImplementedError('HKL calculation is not implemented for CryspyCalculator.')

factory

CalculatorFactory

Factory for creating calculation engine instances.

The factory exposes discovery helpers to list and show available calculators in the current environment and a creator that returns an instantiated calculator or None if the requested one is not available.

Source code in src/easydiffraction/analysis/calculators/factory.py

class CalculatorFactory:
    """Factory for creating calculation engine instances.

    The factory exposes discovery helpers to list and show available
    calculators in the current environment and a creator that returns an
    instantiated calculator or ``None`` if the requested one is not
    available.
    """

    _potential_calculators: Dict[str, Dict[str, Union[str, Type[CalculatorBase]]]] = {
        'crysfml': {
            'description': 'CrysFML library for crystallographic calculations',
            'class': CrysfmlCalculator,
        },
        'cryspy': {
            'description': 'CrysPy library for crystallographic calculations',
            'class': CryspyCalculator,
        },
        'pdffit': {
            'description': 'PDFfit2 library for pair distribution function calculations',
            'class': PdffitCalculator,
        },
    }

    @classmethod
    def _supported_calculators(
        cls,
    ) -> Dict[str, Dict[str, Union[str, Type[CalculatorBase]]]]:
        """Return calculators whose engines are importable.

        This filters the list of potential calculators by instantiating
        their classes and checking the ``engine_imported`` property.

        Returns:
            Mapping from calculator name to its config dict.
        """
        return {
            name: cfg
            for name, cfg in cls._potential_calculators.items()
            if cfg['class']().engine_imported  # instantiate and check the @property
        }

    @classmethod
    def list_supported_calculators(cls) -> List[str]:
        """List names of calculators available in the environment.

        Returns:
            List of calculator identifiers, e.g. ``["crysfml", ...]``.
        """
        return list(cls._supported_calculators().keys())

    @classmethod
    def show_supported_calculators(cls) -> None:
        """Pretty-print supported calculators and their descriptions."""
        columns_headers: List[str] = ['Calculator', 'Description']
        columns_alignment = ['left', 'left']
        columns_data: List[List[str]] = []
        for name, config in cls._supported_calculators().items():
            description: str = config.get('description', 'No description provided.')
            columns_data.append([name, description])

        console.paragraph('Supported calculators')
        render_table(
            columns_headers=columns_headers,
            columns_alignment=columns_alignment,
            columns_data=columns_data,
        )

    @classmethod
    def create_calculator(cls, calculator_name: str) -> Optional[CalculatorBase]:
        """Create a calculator instance by name.

        Args:
            calculator_name: Identifier of the calculator to create.

        Returns:
            A calculator instance or ``None`` if unknown or unsupported.
        """
        config = cls._supported_calculators().get(calculator_name)
        if not config:
            log.warning(
                f"Unknown calculator '{calculator_name}', "
                f'Supported calculators: {cls.list_supported_calculators()}'
            )
            return None

        return config['class']()

create_calculator(calculator_name) classmethod

Create a calculator instance by name.

Parameters:

Name	Type	Description	Default
calculator_name	str	Identifier of the calculator to create.	required

Returns:

Type	Description
Optional[CalculatorBase]	A calculator instance or None if unknown or unsupported.

Source code in src/easydiffraction/analysis/calculators/factory.py

@classmethod
def create_calculator(cls, calculator_name: str) -> Optional[CalculatorBase]:
    """Create a calculator instance by name.

    Args:
        calculator_name: Identifier of the calculator to create.

    Returns:
        A calculator instance or ``None`` if unknown or unsupported.
    """
    config = cls._supported_calculators().get(calculator_name)
    if not config:
        log.warning(
            f"Unknown calculator '{calculator_name}', "
            f'Supported calculators: {cls.list_supported_calculators()}'
        )
        return None

    return config['class']()

list_supported_calculators() classmethod

List names of calculators available in the environment.

Returns:

Type	Description
List[str]	List of calculator identifiers, e.g. ["crysfml", ...].

Source code in src/easydiffraction/analysis/calculators/factory.py

@classmethod
def list_supported_calculators(cls) -> List[str]:
    """List names of calculators available in the environment.

    Returns:
        List of calculator identifiers, e.g. ``["crysfml", ...]``.
    """
    return list(cls._supported_calculators().keys())

show_supported_calculators() classmethod

Pretty-print supported calculators and their descriptions.

Source code in src/easydiffraction/analysis/calculators/factory.py

@classmethod
def show_supported_calculators(cls) -> None:
    """Pretty-print supported calculators and their descriptions."""
    columns_headers: List[str] = ['Calculator', 'Description']
    columns_alignment = ['left', 'left']
    columns_data: List[List[str]] = []
    for name, config in cls._supported_calculators().items():
        description: str = config.get('description', 'No description provided.')
        columns_data.append([name, description])

    console.paragraph('Supported calculators')
    render_table(
        columns_headers=columns_headers,
        columns_alignment=columns_alignment,
        columns_data=columns_data,
    )

pdffit

PDF calculation backend using diffpy.pdffit2 if available.

The class adapts the engine to EasyDiffraction calculator interface and silences stdio on import to avoid noisy output in notebooks and logs.

PdffitCalculator

Bases: CalculatorBase

Wrapper for Pdffit library.

Source code in src/easydiffraction/analysis/calculators/pdffit.py

class PdffitCalculator(CalculatorBase):
    """Wrapper for Pdffit library."""

    engine_imported: bool = PdfFit is not None

    @property
    def name(self):
        return 'pdffit'

    def calculate_structure_factors(self, sample_models, experiments):
        # PDF doesn't compute HKL but we keep interface consistent
        # Intentionally unused, required by public API/signature
        del sample_models, experiments
        print('[pdffit] Calculating HKLs (not applicable)...')
        return []

    def _calculate_single_model_pattern(
        self,
        sample_model: SampleModelBase,
        experiment: ExperimentBase,
        called_by_minimizer: bool = False,
    ):
        # Intentionally unused, required by public API/signature
        del called_by_minimizer

        # Create PDF calculator object
        calculator = PdfFit()

        # ---------------------------
        # Set sample model parameters
        # ---------------------------

        # TODO: move CIF v2 -> CIF v1 conversion to a separate module
        # Convert the sample model to CIF supported by PDFfit
        cif_string_v2 = sample_model.as_cif
        # convert to version 1 of CIF format
        # this means: replace all dots with underscores for
        # cases where the dot is surrounded by letters on both sides.
        pattern = r'(?<=[a-zA-Z])\.(?=[a-zA-Z])'
        cif_string_v1 = re.sub(pattern, '_', cif_string_v2)

        # Create the PDFit structure
        structure = pdffit_cif_parser().parse(cif_string_v1)

        # Set all model parameters:
        # space group, cell parameters, and atom sites (including ADPs)
        calculator.add_structure(structure)

        # -------------------------
        # Set experiment parameters
        # -------------------------

        # Set some peak-related parameters
        calculator.setvar('pscale', experiment.linked_phases[sample_model.name].scale.value)
        calculator.setvar('delta1', experiment.peak.sharp_delta_1.value)
        calculator.setvar('delta2', experiment.peak.sharp_delta_2.value)
        calculator.setvar('spdiameter', experiment.peak.damp_particle_diameter.value)

        # Data
        x = list(experiment.datastore.x)
        y_noise = list(np.zeros_like(x))

        # Assign the data to the PDFfit calculator
        calculator.read_data_lists(
            stype=experiment.type.radiation_probe.value[0].upper(),
            qmax=experiment.peak.cutoff_q.value,
            qdamp=experiment.peak.damp_q.value,
            r_data=x,
            Gr_data=y_noise,
        )

        # qbroad must be set after read_data_lists
        calculator.setvar('qbroad', experiment.peak.broad_q.value)

        # -----------------
        # Calculate pattern
        # -----------------

        # Calculate the PDF pattern
        calculator.calc()

        # Get the calculated PDF pattern
        pattern = calculator.getpdf_fit()
        pattern = np.array(pattern)

        return pattern

categories

aliases

Alias category for mapping friendly names to parameter UIDs.

Defines a small record type used by analysis configuration to refer to parameters via readable labels instead of raw unique identifiers.

Alias

Bases: CategoryItem

Single alias entry.

Maps a human-readable label to a concrete param_uid used by the engine.

Parameters:

Name	Type	Description	Default
label	str	Alias label. Must match ^[A-Za-z_][A-Za-z0-9_]*$.	required
param_uid	str	Target parameter uid. Same identifier pattern as label.	required

Source code in src/easydiffraction/analysis/categories/aliases.py

class Alias(CategoryItem):
    """Single alias entry.

    Maps a human-readable ``label`` to a concrete ``param_uid`` used by
    the engine.

    Args:
        label: Alias label. Must match ``^[A-Za-z_][A-Za-z0-9_]*$``.
        param_uid: Target parameter uid. Same identifier pattern as
            ``label``.
    """

    def __init__(
        self,
        *,
        label: str,
        param_uid: str,
    ) -> None:
        super().__init__()

        self._label: StringDescriptor = StringDescriptor(
            name='label',
            description='...',
            value_spec=AttributeSpec(
                value=label,
                type_=DataTypes.STRING,
                default='...',
                content_validator=RegexValidator(pattern=r'^[A-Za-z_][A-Za-z0-9_]*$'),
            ),
            cif_handler=CifHandler(
                names=[
                    '_alias.label',
                ]
            ),
        )
        self._param_uid: StringDescriptor = StringDescriptor(
            name='param_uid',
            description='...',
            value_spec=AttributeSpec(
                value=param_uid,
                type_=DataTypes.STRING,
                default='...',
                content_validator=RegexValidator(pattern=r'^[A-Za-z_][A-Za-z0-9_]*$'),
            ),
            cif_handler=CifHandler(
                names=[
                    '_alias.param_uid',
                ]
            ),
        )

        self._identity.category_code = 'alias'
        self._identity.category_entry_name = lambda: str(self.label.value)

    @property
    def label(self):
        """Alias label descriptor."""
        return self._label

    @label.setter
    def label(self, value):
        """Set alias label.

        Args:
            value: New label.
        """
        self._label.value = value

    @property
    def param_uid(self):
        """Parameter uid descriptor the alias points to."""
        return self._param_uid

    @param_uid.setter
    def param_uid(self, value):
        """Set the parameter uid.

        Args:
            value: New uid.
        """
        self._param_uid.value = value

label property writable

Alias label descriptor.

param_uid property writable

Parameter uid descriptor the alias points to.

Aliases

Bases: CategoryCollection

Collection of :class:Alias items.

Source code in src/easydiffraction/analysis/categories/aliases.py

class Aliases(CategoryCollection):
    """Collection of :class:`Alias` items."""

    def __init__(self):
        """Create an empty collection of aliases."""
        super().__init__(item_type=Alias)

__init__()

Create an empty collection of aliases.

Source code in src/easydiffraction/analysis/categories/aliases.py

def __init__(self):
    """Create an empty collection of aliases."""
    super().__init__(item_type=Alias)

constraints

Simple symbolic constraint between parameters.

Represents an equation of the form lhs_alias = rhs_expr where rhs_expr is evaluated elsewhere by the analysis engine.

Constraint

Bases: CategoryItem

Single constraint item.

Parameters:

Name	Type	Description	Default
lhs_alias	str	Left-hand side alias name being constrained.	required
rhs_expr	str	Right-hand side expression as a string.	required

Source code in src/easydiffraction/analysis/categories/constraints.py

class Constraint(CategoryItem):
    """Single constraint item.

    Args:
        lhs_alias: Left-hand side alias name being constrained.
        rhs_expr: Right-hand side expression as a string.
    """

    def __init__(
        self,
        *,
        lhs_alias: str,
        rhs_expr: str,
    ) -> None:
        super().__init__()

        self._lhs_alias: StringDescriptor = StringDescriptor(
            name='lhs_alias',
            description='...',
            value_spec=AttributeSpec(
                value=lhs_alias,
                type_=DataTypes.STRING,
                default='...',
                content_validator=RegexValidator(pattern=r'.*'),
            ),
            cif_handler=CifHandler(
                names=[
                    '_constraint.lhs_alias',
                ]
            ),
        )
        self._rhs_expr: StringDescriptor = StringDescriptor(
            name='rhs_expr',
            description='...',
            value_spec=AttributeSpec(
                value=rhs_expr,
                type_=DataTypes.STRING,
                default='...',
                content_validator=RegexValidator(pattern=r'.*'),
            ),
            cif_handler=CifHandler(
                names=[
                    '_constraint.rhs_expr',
                ]
            ),
        )

        self._identity.category_code = 'constraint'
        self._identity.category_entry_name = lambda: str(self.lhs_alias.value)

    @property
    def lhs_alias(self):
        """Alias name on the left-hand side of the equation."""
        return self._lhs_alias

    @lhs_alias.setter
    def lhs_alias(self, value):
        """Set the left-hand side alias.

        Args:
            value: New alias string.
        """
        self._lhs_alias.value = value

    @property
    def rhs_expr(self):
        """Right-hand side expression string."""
        return self._rhs_expr

    @rhs_expr.setter
    def rhs_expr(self, value):
        """Set the right-hand side expression.

        Args:
            value: New expression string.
        """
        self._rhs_expr.value = value

lhs_alias property writable

Alias name on the left-hand side of the equation.

rhs_expr property writable

Right-hand side expression string.

Constraints

Bases: CategoryCollection

Collection of :class:Constraint items.

Source code in src/easydiffraction/analysis/categories/constraints.py

class Constraints(CategoryCollection):
    """Collection of :class:`Constraint` items."""

    def __init__(self):
        """Create an empty constraints collection."""
        super().__init__(item_type=Constraint)

__init__()

Create an empty constraints collection.

Source code in src/easydiffraction/analysis/categories/constraints.py

def __init__(self):
    """Create an empty constraints collection."""
    super().__init__(item_type=Constraint)

joint_fit_experiments

Joint-fit experiment weighting configuration.

Stores per-experiment weights to be used when multiple experiments are fitted simultaneously.

JointFitExperiment

Bases: CategoryItem

A single joint-fit entry.

Parameters:

Name	Type	Description	Default
id	str	Experiment identifier used in the fit session.	required
weight	float	Relative weight factor in the combined objective.	required

Source code in src/easydiffraction/analysis/categories/joint_fit_experiments.py

class JointFitExperiment(CategoryItem):
    """A single joint-fit entry.

    Args:
        id: Experiment identifier used in the fit session.
        weight: Relative weight factor in the combined objective.
    """

    def __init__(
        self,
        *,
        id: str,
        weight: float,
    ) -> None:
        super().__init__()

        self._id: StringDescriptor = StringDescriptor(
            name='id',  # TODO: need new name instead of id
            description='...',
            value_spec=AttributeSpec(
                value=id,
                type_=DataTypes.STRING,
                default='...',
                content_validator=RegexValidator(pattern=r'^[A-Za-z_][A-Za-z0-9_]*$'),
            ),
            cif_handler=CifHandler(
                names=[
                    '_joint_fit_experiment.id',
                ]
            ),
        )
        self._weight: NumericDescriptor = NumericDescriptor(
            name='weight',
            description='...',
            value_spec=AttributeSpec(
                value=weight,
                type_=DataTypes.NUMERIC,
                default=0.0,
                content_validator=RangeValidator(),
            ),
            cif_handler=CifHandler(
                names=[
                    '_joint_fit_experiment.weight',
                ]
            ),
        )

        self._identity.category_code = 'joint_fit_experiment'
        self._identity.category_entry_name = lambda: str(self.id.value)

    @property
    def id(self):
        """Experiment identifier descriptor."""
        return self._id

    @id.setter
    def id(self, value):
        """Set the experiment identifier.

        Args:
            value: New id string.
        """
        self._id.value = value

    @property
    def weight(self):
        """Weight factor descriptor."""
        return self._weight

    @weight.setter
    def weight(self, value):
        """Set the weight factor.

        Args:
            value: New weight value.
        """
        self._weight.value = value

id property writable

Experiment identifier descriptor.

weight property writable

Weight factor descriptor.

JointFitExperiments

Bases: CategoryCollection

Collection of :class:JointFitExperiment items.

Source code in src/easydiffraction/analysis/categories/joint_fit_experiments.py

class JointFitExperiments(CategoryCollection):
    """Collection of :class:`JointFitExperiment` items."""

    def __init__(self):
        """Create an empty joint-fit experiments collection."""
        super().__init__(item_type=JointFitExperiment)

__init__()

Create an empty joint-fit experiments collection.

Source code in src/easydiffraction/analysis/categories/joint_fit_experiments.py

def __init__(self):
    """Create an empty joint-fit experiments collection."""
    super().__init__(item_type=JointFitExperiment)

fit_helpers

metrics

calculate_r_factor(y_obs, y_calc)

Calculate the R-factor (reliability factor) between observed and calculated data.

Parameters:

Name	Type	Description	Default
y_obs	ndarray	Observed data points.	required
y_calc	ndarray	Calculated data points.	required

Returns:

Type	Description
float	R-factor value.

Source code in src/easydiffraction/analysis/fit_helpers/metrics.py

def calculate_r_factor(
    y_obs: np.ndarray,
    y_calc: np.ndarray,
) -> float:
    """Calculate the R-factor (reliability factor) between observed and
    calculated data.

    Args:
        y_obs: Observed data points.
        y_calc: Calculated data points.

    Returns:
        R-factor value.
    """
    y_obs = np.asarray(y_obs)
    y_calc = np.asarray(y_calc)
    numerator = np.sum(np.abs(y_obs - y_calc))
    denominator = np.sum(np.abs(y_obs))
    return numerator / denominator if denominator != 0 else np.nan

calculate_r_factor_squared(y_obs, y_calc)

Calculate the R-factor squared between observed and calculated data.

Parameters:

Name	Type	Description	Default
y_obs	ndarray	Observed data points.	required
y_calc	ndarray	Calculated data points.	required

Returns:

Type	Description
float	R-factor squared value.

Source code in src/easydiffraction/analysis/fit_helpers/metrics.py

def calculate_r_factor_squared(
    y_obs: np.ndarray,
    y_calc: np.ndarray,
) -> float:
    """Calculate the R-factor squared between observed and calculated
    data.

    Args:
        y_obs: Observed data points.
        y_calc: Calculated data points.

    Returns:
        R-factor squared value.
    """
    y_obs = np.asarray(y_obs)
    y_calc = np.asarray(y_calc)
    numerator = np.sum((y_obs - y_calc) ** 2)
    denominator = np.sum(y_obs**2)
    return np.sqrt(numerator / denominator) if denominator != 0 else np.nan

calculate_rb_factor(y_obs, y_calc)

Calculate the Bragg R-factor between observed and calculated data.

Parameters:

Name	Type	Description	Default
y_obs	ndarray	Observed data points.	required
y_calc	ndarray	Calculated data points.	required

Returns:

Type	Description
float	Bragg R-factor value.

Source code in src/easydiffraction/analysis/fit_helpers/metrics.py

def calculate_rb_factor(
    y_obs: np.ndarray,
    y_calc: np.ndarray,
) -> float:
    """Calculate the Bragg R-factor between observed and calculated
    data.

    Args:
        y_obs: Observed data points.
        y_calc: Calculated data points.

    Returns:
        Bragg R-factor value.
    """
    y_obs = np.asarray(y_obs)
    y_calc = np.asarray(y_calc)
    numerator = np.sum(np.abs(y_obs - y_calc))
    denominator = np.sum(y_obs)
    return numerator / denominator if denominator != 0 else np.nan

calculate_reduced_chi_square(residuals, num_parameters)

Calculate the reduced chi-square statistic.

Parameters:

Name	Type	Description	Default
residuals	ndarray	Residuals between observed and calculated data.	required
num_parameters	int	Number of free parameters used in the model.	required

Returns:

Type	Description
float	Reduced chi-square value.

Source code in src/easydiffraction/analysis/fit_helpers/metrics.py

def calculate_reduced_chi_square(
    residuals: np.ndarray,
    num_parameters: int,
) -> float:
    """Calculate the reduced chi-square statistic.

    Args:
        residuals: Residuals between observed and calculated data.
        num_parameters: Number of free parameters used in the model.

    Returns:
        Reduced chi-square value.
    """
    residuals = np.asarray(residuals)
    chi_square = np.sum(residuals**2)
    n_points = len(residuals)
    dof = n_points - num_parameters
    if dof > 0:
        return chi_square / dof
    else:
        return np.nan

calculate_weighted_r_factor(y_obs, y_calc, weights)

Calculate the weighted R-factor between observed and calculated data.

Parameters:

Name	Type	Description	Default
y_obs	ndarray	Observed data points.	required
y_calc	ndarray	Calculated data points.	required
weights	ndarray	Weights for each data point.	required

Returns:

Type	Description
float	Weighted R-factor value.

Source code in src/easydiffraction/analysis/fit_helpers/metrics.py

def calculate_weighted_r_factor(
    y_obs: np.ndarray,
    y_calc: np.ndarray,
    weights: np.ndarray,
) -> float:
    """Calculate the weighted R-factor between observed and calculated
    data.

    Args:
        y_obs: Observed data points.
        y_calc: Calculated data points.
        weights: Weights for each data point.

    Returns:
        Weighted R-factor value.
    """
    y_obs = np.asarray(y_obs)
    y_calc = np.asarray(y_calc)
    weights = np.asarray(weights)
    numerator = np.sum(weights * (y_obs - y_calc) ** 2)
    denominator = np.sum(weights * y_obs**2)
    return np.sqrt(numerator / denominator) if denominator != 0 else np.nan

get_reliability_inputs(sample_models, experiments, calculator)

Collect observed and calculated data points for reliability calculations.

Parameters:

Name	Type	Description	Default
sample_models	SampleModels	Collection of sample models.	required
experiments	Experiments	Collection of experiments.	required
calculator	CalculatorBase	The calculator to use for pattern generation.	required

Returns:

Type	Description
Tuple[ndarray, ndarray, Optional[ndarray]]	Tuple containing arrays of (observed values, calculated values, error values)

Source code in src/easydiffraction/analysis/fit_helpers/metrics.py

def get_reliability_inputs(
    sample_models: SampleModels,
    experiments: Experiments,
    calculator: CalculatorBase,
) -> Tuple[np.ndarray, np.ndarray, Optional[np.ndarray]]:
    """Collect observed and calculated data points for reliability
    calculations.

    Args:
        sample_models: Collection of sample models.
        experiments: Collection of experiments.
        calculator: The calculator to use for pattern generation.

    Returns:
        Tuple containing arrays of (observed values, calculated values,
            error values)
    """
    y_obs_all = []
    y_calc_all = []
    y_err_all = []
    for experiment in experiments.values():
        calculator.calculate_pattern(sample_models, experiment)
        y_calc = experiment.datastore.calc
        y_meas = experiment.datastore.meas
        y_meas_su = experiment.datastore.meas_su

        if y_meas is not None and y_calc is not None:
            # If standard uncertainty is not provided, use ones
            if y_meas_su is None:
                y_meas_su = np.ones_like(y_meas)

            y_obs_all.extend(y_meas)
            y_calc_all.extend(y_calc)
            y_err_all.extend(y_meas_su)

    return (
        np.array(y_obs_all),
        np.array(y_calc_all),
        np.array(y_err_all) if y_err_all else None,
    )

reporting

FitResults

Container for results of a single optimization run.

Holds success flag, chi-square metrics, iteration counts, timing, and parameter objects. Provides a printer to summarize key indicators and a table of fitted parameters.

Source code in src/easydiffraction/analysis/fit_helpers/reporting.py

class FitResults:
    """Container for results of a single optimization run.

    Holds success flag, chi-square metrics, iteration counts, timing,
    and parameter objects. Provides a printer to summarize key
    indicators and a table of fitted parameters.
    """

    def __init__(
        self,
        success: bool = False,
        parameters: Optional[List[Any]] = None,
        chi_square: Optional[float] = None,
        reduced_chi_square: Optional[float] = None,
        message: str = '',
        iterations: int = 0,
        engine_result: Optional[Any] = None,
        starting_parameters: Optional[List[Any]] = None,
        fitting_time: Optional[float] = None,
        **kwargs: Any,
    ) -> None:
        """Initialize FitResults with the given parameters.

        Args:
            success: Indicates if the fit was successful.
            parameters: List of parameters used in the fit.
            chi_square: Chi-square value of the fit.
            reduced_chi_square: Reduced chi-square value of the fit.
            message: Message related to the fit.
            iterations: Number of iterations performed.
            engine_result: Result from the fitting engine.
            starting_parameters: Initial parameters for the fit.
            fitting_time: Time taken for the fitting process.
            **kwargs: Additional engine-specific fields. If ``redchi``
                is provided and ``reduced_chi_square`` is not set, it is
                used as the reduced chi-square value.
        """
        self.success: bool = success
        self.parameters: List[Any] = parameters if parameters is not None else []
        self.chi_square: Optional[float] = chi_square
        self.reduced_chi_square: Optional[float] = reduced_chi_square
        self.message: str = message
        self.iterations: int = iterations
        self.engine_result: Optional[Any] = engine_result
        self.result: Optional[Any] = None
        self.starting_parameters: List[Any] = (
            starting_parameters if starting_parameters is not None else []
        )
        self.fitting_time: Optional[float] = fitting_time

        if 'redchi' in kwargs and self.reduced_chi_square is None:
            self.reduced_chi_square = kwargs.get('redchi')

        for key, value in kwargs.items():
            setattr(self, key, value)

    def display_results(
        self,
        y_obs: Optional[List[float]] = None,
        y_calc: Optional[List[float]] = None,
        y_err: Optional[List[float]] = None,
        f_obs: Optional[List[float]] = None,
        f_calc: Optional[List[float]] = None,
    ) -> None:
        """Render a human-readable summary of the fit.

        Args:
            y_obs: Observed intensities for pattern R-factor metrics.
            y_calc: Calculated intensities for pattern R-factor metrics.
            y_err: Standard deviations of observed intensities for wR.
            f_obs: Observed structure-factor magnitudes for Bragg R.
            f_calc: Calculated structure-factor magnitudes for Bragg R.
        """
        status_icon = '✅' if self.success else '❌'
        rf = rf2 = wr = br = None
        if y_obs is not None and y_calc is not None:
            rf = calculate_r_factor(y_obs, y_calc) * 100
            rf2 = calculate_r_factor_squared(y_obs, y_calc) * 100
        if y_obs is not None and y_calc is not None and y_err is not None:
            wr = calculate_weighted_r_factor(y_obs, y_calc, y_err) * 100
        if f_obs is not None and f_calc is not None:
            br = calculate_rb_factor(f_obs, f_calc) * 100

        console.paragraph('Fit results')
        console.print(f'{status_icon} Success: {self.success}')
        console.print(f'⏱️ Fitting time: {self.fitting_time:.2f} seconds')
        console.print(f'📏 Goodness-of-fit (reduced χ²): {self.reduced_chi_square:.2f}')
        if rf is not None:
            console.print(f'📏 R-factor (Rf): {rf:.2f}%')
        if rf2 is not None:
            console.print(f'📏 R-factor squared (Rf²): {rf2:.2f}%')
        if wr is not None:
            console.print(f'📏 Weighted R-factor (wR): {wr:.2f}%')
        if br is not None:
            console.print(f'📏 Bragg R-factor (BR): {br:.2f}%')
        console.print('📈 Fitted parameters:')

        headers = [
            'datablock',
            'category',
            'entry',
            'parameter',
            'start',
            'fitted',
            'uncertainty',
            'units',
            'change',
        ]
        alignments = [
            'left',
            'left',
            'left',
            'left',
            'right',
            'right',
            'right',
            'left',
            'right',
        ]

        rows = []
        for param in self.parameters:
            datablock_entry_name = (
                param._identity.datablock_entry_name
            )  # getattr(param, 'datablock_name', 'N/A')
            category_code = param._identity.category_code  # getattr(param, 'category_key', 'N/A')
            category_entry_name = (
                param._identity.category_entry_name
            )  # getattr(param, 'category_entry_name', 'N/A')
            name = getattr(param, 'name', 'N/A')
            start = (
                f'{getattr(param, "_fit_start_value", "N/A"):.4f}'
                if param._fit_start_value is not None
                else 'N/A'
            )
            fitted = f'{param.value:.4f}' if param.value is not None else 'N/A'
            uncertainty = f'{param.uncertainty:.4f}' if param.uncertainty is not None else 'N/A'
            units = getattr(param, 'units', 'N/A')

            if param._fit_start_value and param.value:
                change = ((param.value - param._fit_start_value) / param._fit_start_value) * 100
                arrow = '↑' if change > 0 else '↓'
                relative_change = f'{abs(change):.2f} % {arrow}'
            else:
                relative_change = 'N/A'

            rows.append([
                datablock_entry_name,
                category_code,
                category_entry_name,
                name,
                start,
                fitted,
                uncertainty,
                units,
                relative_change,
            ])

        render_table(
            columns_headers=headers,
            columns_alignment=alignments,
            columns_data=rows,
        )

__init__(success=False, parameters=None, chi_square=None, reduced_chi_square=None, message='', iterations=0, engine_result=None, starting_parameters=None, fitting_time=None, **kwargs)

Initialize FitResults with the given parameters.

Parameters:

Name	Type	Description	Default
success	bool	Indicates if the fit was successful.	False
parameters	Optional[List[Any]]	List of parameters used in the fit.	None
chi_square	Optional[float]	Chi-square value of the fit.	None
reduced_chi_square	Optional[float]	Reduced chi-square value of the fit.	None
message	str	Message related to the fit.	''
iterations	int	Number of iterations performed.	0
engine_result	Optional[Any]	Result from the fitting engine.	None
starting_parameters	Optional[List[Any]]	Initial parameters for the fit.	None
fitting_time	Optional[float]	Time taken for the fitting process.	None
**kwargs	Any	Additional engine-specific fields. If redchi is provided and reduced_chi_square is not set, it is used as the reduced chi-square value.	{}

Source code in src/easydiffraction/analysis/fit_helpers/reporting.py

def __init__(
    self,
    success: bool = False,
    parameters: Optional[List[Any]] = None,
    chi_square: Optional[float] = None,
    reduced_chi_square: Optional[float] = None,
    message: str = '',
    iterations: int = 0,
    engine_result: Optional[Any] = None,
    starting_parameters: Optional[List[Any]] = None,
    fitting_time: Optional[float] = None,
    **kwargs: Any,
) -> None:
    """Initialize FitResults with the given parameters.

    Args:
        success: Indicates if the fit was successful.
        parameters: List of parameters used in the fit.
        chi_square: Chi-square value of the fit.
        reduced_chi_square: Reduced chi-square value of the fit.
        message: Message related to the fit.
        iterations: Number of iterations performed.
        engine_result: Result from the fitting engine.
        starting_parameters: Initial parameters for the fit.
        fitting_time: Time taken for the fitting process.
        **kwargs: Additional engine-specific fields. If ``redchi``
            is provided and ``reduced_chi_square`` is not set, it is
            used as the reduced chi-square value.
    """
    self.success: bool = success
    self.parameters: List[Any] = parameters if parameters is not None else []
    self.chi_square: Optional[float] = chi_square
    self.reduced_chi_square: Optional[float] = reduced_chi_square
    self.message: str = message
    self.iterations: int = iterations
    self.engine_result: Optional[Any] = engine_result
    self.result: Optional[Any] = None
    self.starting_parameters: List[Any] = (
        starting_parameters if starting_parameters is not None else []
    )
    self.fitting_time: Optional[float] = fitting_time

    if 'redchi' in kwargs and self.reduced_chi_square is None:
        self.reduced_chi_square = kwargs.get('redchi')

    for key, value in kwargs.items():
        setattr(self, key, value)

display_results(y_obs=None, y_calc=None, y_err=None, f_obs=None, f_calc=None)

Render a human-readable summary of the fit.

Parameters:

Name	Type	Description	Default
y_obs	Optional[List[float]]	Observed intensities for pattern R-factor metrics.	None
y_calc	Optional[List[float]]	Calculated intensities for pattern R-factor metrics.	None
y_err	Optional[List[float]]	Standard deviations of observed intensities for wR.	None
f_obs	Optional[List[float]]	Observed structure-factor magnitudes for Bragg R.	None
f_calc	Optional[List[float]]	Calculated structure-factor magnitudes for Bragg R.	None

Source code in src/easydiffraction/analysis/fit_helpers/reporting.py

def display_results(
    self,
    y_obs: Optional[List[float]] = None,
    y_calc: Optional[List[float]] = None,
    y_err: Optional[List[float]] = None,
    f_obs: Optional[List[float]] = None,
    f_calc: Optional[List[float]] = None,
) -> None:
    """Render a human-readable summary of the fit.

    Args:
        y_obs: Observed intensities for pattern R-factor metrics.
        y_calc: Calculated intensities for pattern R-factor metrics.
        y_err: Standard deviations of observed intensities for wR.
        f_obs: Observed structure-factor magnitudes for Bragg R.
        f_calc: Calculated structure-factor magnitudes for Bragg R.
    """
    status_icon = '✅' if self.success else '❌'
    rf = rf2 = wr = br = None
    if y_obs is not None and y_calc is not None:
        rf = calculate_r_factor(y_obs, y_calc) * 100
        rf2 = calculate_r_factor_squared(y_obs, y_calc) * 100
    if y_obs is not None and y_calc is not None and y_err is not None:
        wr = calculate_weighted_r_factor(y_obs, y_calc, y_err) * 100
    if f_obs is not None and f_calc is not None:
        br = calculate_rb_factor(f_obs, f_calc) * 100

    console.paragraph('Fit results')
    console.print(f'{status_icon} Success: {self.success}')
    console.print(f'⏱️ Fitting time: {self.fitting_time:.2f} seconds')
    console.print(f'📏 Goodness-of-fit (reduced χ²): {self.reduced_chi_square:.2f}')
    if rf is not None:
        console.print(f'📏 R-factor (Rf): {rf:.2f}%')
    if rf2 is not None:
        console.print(f'📏 R-factor squared (Rf²): {rf2:.2f}%')
    if wr is not None:
        console.print(f'📏 Weighted R-factor (wR): {wr:.2f}%')
    if br is not None:
        console.print(f'📏 Bragg R-factor (BR): {br:.2f}%')
    console.print('📈 Fitted parameters:')

    headers = [
        'datablock',
        'category',
        'entry',
        'parameter',
        'start',
        'fitted',
        'uncertainty',
        'units',
        'change',
    ]
    alignments = [
        'left',
        'left',
        'left',
        'left',
        'right',
        'right',
        'right',
        'left',
        'right',
    ]

    rows = []
    for param in self.parameters:
        datablock_entry_name = (
            param._identity.datablock_entry_name
        )  # getattr(param, 'datablock_name', 'N/A')
        category_code = param._identity.category_code  # getattr(param, 'category_key', 'N/A')
        category_entry_name = (
            param._identity.category_entry_name
        )  # getattr(param, 'category_entry_name', 'N/A')
        name = getattr(param, 'name', 'N/A')
        start = (
            f'{getattr(param, "_fit_start_value", "N/A"):.4f}'
            if param._fit_start_value is not None
            else 'N/A'
        )
        fitted = f'{param.value:.4f}' if param.value is not None else 'N/A'
        uncertainty = f'{param.uncertainty:.4f}' if param.uncertainty is not None else 'N/A'
        units = getattr(param, 'units', 'N/A')

        if param._fit_start_value and param.value:
            change = ((param.value - param._fit_start_value) / param._fit_start_value) * 100
            arrow = '↑' if change > 0 else '↓'
            relative_change = f'{abs(change):.2f} % {arrow}'
        else:
            relative_change = 'N/A'

        rows.append([
            datablock_entry_name,
            category_code,
            category_entry_name,
            name,
            start,
            fitted,
            uncertainty,
            units,
            relative_change,
        ])

    render_table(
        columns_headers=headers,
        columns_alignment=alignments,
        columns_data=rows,
    )

tracking

FitProgressTracker

Track and report reduced chi-square during optimization.

The tracker keeps iteration counters, remembers the best observed reduced chi-square and when it occurred, and can display progress as a table in notebooks or a text UI in terminals.

Source code in src/easydiffraction/analysis/fit_helpers/tracking.py

class FitProgressTracker:
    """Track and report reduced chi-square during optimization.

    The tracker keeps iteration counters, remembers the best observed
    reduced chi-square and when it occurred, and can display progress as
    a table in notebooks or a text UI in terminals.
    """

    def __init__(self) -> None:
        self._iteration: int = 0
        self._previous_chi2: Optional[float] = None
        self._last_chi2: Optional[float] = None
        self._last_iteration: Optional[int] = None
        self._best_chi2: Optional[float] = None
        self._best_iteration: Optional[int] = None
        self._fitting_time: Optional[float] = None

        self._df_rows: List[List[str]] = []
        self._display_handle: Optional[Any] = None
        self._live: Optional[Any] = None

    def reset(self) -> None:
        """Reset internal state before a new optimization run."""
        self._iteration = 0
        self._previous_chi2 = None
        self._last_chi2 = None
        self._last_iteration = None
        self._best_chi2 = None
        self._best_iteration = None
        self._fitting_time = None

    def track(
        self,
        residuals: np.ndarray,
        parameters: List[float],
    ) -> np.ndarray:
        """Update progress with current residuals and parameters.

        Args:
            residuals: Residuals between measured and calculated data.
            parameters: Current free parameters being fitted.

        Returns:
            Residuals unchanged, for optimizer consumption.
        """
        self._iteration += 1

        reduced_chi2 = calculate_reduced_chi_square(residuals, len(parameters))

        row: List[str] = []

        # First iteration, initialize tracking
        if self._previous_chi2 is None:
            self._previous_chi2 = reduced_chi2
            self._best_chi2 = reduced_chi2
            self._best_iteration = self._iteration

            row = [
                str(self._iteration),
                f'{reduced_chi2:.2f}',
                '',
            ]

        # Subsequent iterations, check for significant changes
        else:
            change = (self._previous_chi2 - reduced_chi2) / self._previous_chi2

            # Improvement check
            if change > SIGNIFICANT_CHANGE_THRESHOLD:
                change_in_percent = change * 100

                row = [
                    str(self._iteration),
                    f'{reduced_chi2:.2f}',
                    f'{change_in_percent:.1f}% ↓',
                ]

                self._previous_chi2 = reduced_chi2

        # Output if there is something new to display
        if row:
            self.add_tracking_info(row)

        # Update best chi-square if better
        if reduced_chi2 < self._best_chi2:
            self._best_chi2 = reduced_chi2
            self._best_iteration = self._iteration

        # Store last chi-square and iteration
        self._last_chi2 = reduced_chi2
        self._last_iteration = self._iteration

        return residuals

    @property
    def best_chi2(self) -> Optional[float]:
        """Best recorded reduced chi-square value or None."""
        return self._best_chi2

    @property
    def best_iteration(self) -> Optional[int]:
        """Iteration index at which the best chi-square was observed."""
        return self._best_iteration

    @property
    def iteration(self) -> int:
        """Current iteration counter."""
        return self._iteration

    @property
    def fitting_time(self) -> Optional[float]:
        """Elapsed time of the last run in seconds, if available."""
        return self._fitting_time

    def start_timer(self) -> None:
        """Begin timing of a fit run."""
        self._start_time = time.perf_counter()

    def stop_timer(self) -> None:
        """Stop timing and store elapsed time for the run."""
        self._end_time = time.perf_counter()
        self._fitting_time = self._end_time - self._start_time

    def start_tracking(self, minimizer_name: str) -> None:
        """Initialize display and headers and announce the minimizer.

        Args:
            minimizer_name: Name of the minimizer used for the run.
        """
        console.print(f"🚀 Starting fit process with '{minimizer_name}'...")
        console.print('📈 Goodness-of-fit (reduced χ²) change:')

        # Reset rows and create an environment-appropriate handle
        self._df_rows = []
        self._display_handle = _make_display_handle()

        # Initial empty table; subsequent updates will reuse the handle
        render_table(
            columns_headers=DEFAULT_HEADERS,
            columns_alignment=DEFAULT_ALIGNMENTS,
            columns_data=self._df_rows,
            display_handle=self._display_handle,
        )

    def add_tracking_info(self, row: List[str]) -> None:
        """Append a formatted row to the progress display.

        Args:
            row: Columns corresponding to DEFAULT_HEADERS.
        """
        # Append and update via the active handle (Jupyter or
        # terminal live)
        self._df_rows.append(row)
        render_table(
            columns_headers=DEFAULT_HEADERS,
            columns_alignment=DEFAULT_ALIGNMENTS,
            columns_data=self._df_rows,
            display_handle=self._display_handle,
        )

    def finish_tracking(self) -> None:
        """Finalize progress display and print best result summary."""
        # Add last iteration as last row
        row: List[str] = [
            str(self._last_iteration),
            f'{self._last_chi2:.2f}' if self._last_chi2 is not None else '',
            '',
        ]
        self.add_tracking_info(row)

        # Close terminal live if used
        if self._display_handle is not None and hasattr(self._display_handle, 'close'):
            with suppress(Exception):
                self._display_handle.close()

        # Print best result
        console.print(
            f'🏆 Best goodness-of-fit (reduced χ²) is {self._best_chi2:.2f} '
            f'at iteration {self._best_iteration}'
        )
        console.print('✅ Fitting complete.')

add_tracking_info(row)

Append a formatted row to the progress display.

Parameters:

Name	Type	Description	Default
row	List[str]	Columns corresponding to DEFAULT_HEADERS.	required

Source code in src/easydiffraction/analysis/fit_helpers/tracking.py

def add_tracking_info(self, row: List[str]) -> None:
    """Append a formatted row to the progress display.

    Args:
        row: Columns corresponding to DEFAULT_HEADERS.
    """
    # Append and update via the active handle (Jupyter or
    # terminal live)
    self._df_rows.append(row)
    render_table(
        columns_headers=DEFAULT_HEADERS,
        columns_alignment=DEFAULT_ALIGNMENTS,
        columns_data=self._df_rows,
        display_handle=self._display_handle,
    )

best_chi2 property

Best recorded reduced chi-square value or None.

best_iteration property

Iteration index at which the best chi-square was observed.

finish_tracking()

Finalize progress display and print best result summary.

Source code in src/easydiffraction/analysis/fit_helpers/tracking.py

def finish_tracking(self) -> None:
    """Finalize progress display and print best result summary."""
    # Add last iteration as last row
    row: List[str] = [
        str(self._last_iteration),
        f'{self._last_chi2:.2f}' if self._last_chi2 is not None else '',
        '',
    ]
    self.add_tracking_info(row)

    # Close terminal live if used
    if self._display_handle is not None and hasattr(self._display_handle, 'close'):
        with suppress(Exception):
            self._display_handle.close()

    # Print best result
    console.print(
        f'🏆 Best goodness-of-fit (reduced χ²) is {self._best_chi2:.2f} '
        f'at iteration {self._best_iteration}'
    )
    console.print('✅ Fitting complete.')

fitting_time property

Elapsed time of the last run in seconds, if available.

iteration property

Current iteration counter.

reset()

Reset internal state before a new optimization run.

Source code in src/easydiffraction/analysis/fit_helpers/tracking.py

def reset(self) -> None:
    """Reset internal state before a new optimization run."""
    self._iteration = 0
    self._previous_chi2 = None
    self._last_chi2 = None
    self._last_iteration = None
    self._best_chi2 = None
    self._best_iteration = None
    self._fitting_time = None

start_timer()

Begin timing of a fit run.

Source code in src/easydiffraction/analysis/fit_helpers/tracking.py

def start_timer(self) -> None:
    """Begin timing of a fit run."""
    self._start_time = time.perf_counter()

start_tracking(minimizer_name)

Initialize display and headers and announce the minimizer.

Parameters:

Name	Type	Description	Default
minimizer_name	str	Name of the minimizer used for the run.	required

Source code in src/easydiffraction/analysis/fit_helpers/tracking.py

def start_tracking(self, minimizer_name: str) -> None:
    """Initialize display and headers and announce the minimizer.

    Args:
        minimizer_name: Name of the minimizer used for the run.
    """
    console.print(f"🚀 Starting fit process with '{minimizer_name}'...")
    console.print('📈 Goodness-of-fit (reduced χ²) change:')

    # Reset rows and create an environment-appropriate handle
    self._df_rows = []
    self._display_handle = _make_display_handle()

    # Initial empty table; subsequent updates will reuse the handle
    render_table(
        columns_headers=DEFAULT_HEADERS,
        columns_alignment=DEFAULT_ALIGNMENTS,
        columns_data=self._df_rows,
        display_handle=self._display_handle,
    )

stop_timer()

Stop timing and store elapsed time for the run.

Source code in src/easydiffraction/analysis/fit_helpers/tracking.py

def stop_timer(self) -> None:
    """Stop timing and store elapsed time for the run."""
    self._end_time = time.perf_counter()
    self._fitting_time = self._end_time - self._start_time

track(residuals, parameters)

Update progress with current residuals and parameters.

Parameters:

Name	Type	Description	Default
residuals	ndarray	Residuals between measured and calculated data.	required
parameters	List[float]	Current free parameters being fitted.	required

Returns:

Type	Description
ndarray	Residuals unchanged, for optimizer consumption.

Source code in src/easydiffraction/analysis/fit_helpers/tracking.py

def track(
    self,
    residuals: np.ndarray,
    parameters: List[float],
) -> np.ndarray:
    """Update progress with current residuals and parameters.

    Args:
        residuals: Residuals between measured and calculated data.
        parameters: Current free parameters being fitted.

    Returns:
        Residuals unchanged, for optimizer consumption.
    """
    self._iteration += 1

    reduced_chi2 = calculate_reduced_chi_square(residuals, len(parameters))

    row: List[str] = []

    # First iteration, initialize tracking
    if self._previous_chi2 is None:
        self._previous_chi2 = reduced_chi2
        self._best_chi2 = reduced_chi2
        self._best_iteration = self._iteration

        row = [
            str(self._iteration),
            f'{reduced_chi2:.2f}',
            '',
        ]

    # Subsequent iterations, check for significant changes
    else:
        change = (self._previous_chi2 - reduced_chi2) / self._previous_chi2

        # Improvement check
        if change > SIGNIFICANT_CHANGE_THRESHOLD:
            change_in_percent = change * 100

            row = [
                str(self._iteration),
                f'{reduced_chi2:.2f}',
                f'{change_in_percent:.1f}% ↓',
            ]

            self._previous_chi2 = reduced_chi2

    # Output if there is something new to display
    if row:
        self.add_tracking_info(row)

    # Update best chi-square if better
    if reduced_chi2 < self._best_chi2:
        self._best_chi2 = reduced_chi2
        self._best_iteration = self._iteration

    # Store last chi-square and iteration
    self._last_chi2 = reduced_chi2
    self._last_iteration = self._iteration

    return residuals

fitting

Fitter

Handles the fitting workflow using a pluggable minimizer.

Source code in src/easydiffraction/analysis/fitting.py

class Fitter:
    """Handles the fitting workflow using a pluggable minimizer."""

    def __init__(self, selection: str = 'lmfit (leastsq)') -> None:
        self.selection: str = selection
        self.engine: str = selection.split(' ')[0]  # Extracts 'lmfit' or 'dfols'
        self.minimizer = MinimizerFactory.create_minimizer(selection)
        self.results: Optional[FitResults] = None

    def fit(
        self,
        sample_models: SampleModels,
        experiments: Experiments,
        calculator: Any,
        weights: Optional[np.array] = None,
    ) -> None:
        """Run the fitting process.

        Args:
            sample_models: Collection of sample models.
            experiments: Collection of experiments.
            calculator: The calculator to use for pattern generation.
            weights: Optional weights for joint fitting.
        """
        params = sample_models.free_parameters + experiments.free_parameters

        if not params:
            print('⚠️ No parameters selected for fitting.')
            return None

        for param in params:
            param._fit_start_value = param.value

        def objective_function(engine_params: Dict[str, Any]) -> np.ndarray:
            return self._residual_function(
                engine_params=engine_params,
                parameters=params,
                sample_models=sample_models,
                experiments=experiments,
                calculator=calculator,
                weights=weights,
            )

        # Perform fitting
        self.results = self.minimizer.fit(params, objective_function)

        # Post-fit processing
        self._process_fit_results(sample_models, experiments, calculator)

    def _process_fit_results(
        self,
        sample_models: SampleModels,
        experiments: Experiments,
        calculator: CalculatorBase,
    ) -> None:
        """Collect reliability inputs and display results after fitting.

        Args:
            sample_models: Collection of sample models.
            experiments: Collection of experiments.
            calculator: The calculator used for pattern generation.
        """
        y_obs, y_calc, y_err = get_reliability_inputs(
            sample_models,
            experiments,
            calculator,
        )

        # Placeholder for future f_obs / f_calc retrieval
        f_obs, f_calc = None, None

        if self.results:
            self.results.display_results(
                y_obs=y_obs,
                y_calc=y_calc,
                y_err=y_err,
                f_obs=f_obs,
                f_calc=f_calc,
            )

    def _collect_free_parameters(
        self,
        sample_models: SampleModels,
        experiments: Experiments,
    ) -> List[Parameter]:
        """Collect free parameters from sample models and experiments.

        Args:
            sample_models: Collection of sample models.
            experiments: Collection of experiments.

        Returns:
            List of free parameters.
        """
        free_params: List[Parameter] = sample_models.free_parameters + experiments.free_parameters
        return free_params

    def _residual_function(
        self,
        engine_params: Dict[str, Any],
        parameters: List[Parameter],
        sample_models: SampleModels,
        experiments: Experiments,
        calculator: CalculatorBase,
        weights: Optional[np.array] = None,
    ) -> np.ndarray:
        """Residual function computes the difference between measured
        and calculated patterns. It updates the parameter values
        according to the optimizer-provided engine_params.

        Args:
            engine_params: Engine-specific parameter dict.
            parameters: List of parameters being optimized.
            sample_models: Collection of sample models.
            experiments: Collection of experiments.
            calculator: The calculator to use for pattern generation.
            weights: Optional weights for joint fitting.

        Returns:
            Array of weighted residuals.
        """
        # Sync parameters back to objects
        self.minimizer._sync_result_to_parameters(parameters, engine_params)

        # Prepare weights for joint fitting
        num_expts: int = len(experiments.names)
        if weights is None:
            _weights = np.ones(num_expts)
        else:
            _weights_list: List[float] = []
            for name in experiments.names:
                _weight = weights[name].weight.value
                _weights_list.append(_weight)
            _weights = np.array(_weights_list, dtype=np.float64)

        # Normalize weights so they sum to num_expts
        # We should obtain the same reduced chi_squared when a single
        # dataset is split into two parts and fit together. If weights
        # sum to one, then reduced chi_squared will be half as large as
        # expected.
        _weights *= num_expts / np.sum(_weights)
        residuals: List[float] = []

        for experiment, weight in zip(experiments.values(), _weights, strict=True):
            # Calculate the difference between measured and calculated
            # patterns
            calculator.calculate_pattern(
                sample_models,
                experiment,
                called_by_minimizer=True,
            )
            y_calc: np.ndarray = experiment.datastore.calc
            y_meas: np.ndarray = experiment.datastore.meas
            y_meas_su: np.ndarray = experiment.datastore.meas_su
            diff = (y_meas - y_calc) / y_meas_su

            # Residuals are squared before going into reduced
            # chi-squared
            diff *= np.sqrt(weight)

            # Append the residuals for this experiment
            residuals.extend(diff)

        return self.minimizer.tracker.track(np.array(residuals), parameters)

fit(sample_models, experiments, calculator, weights=None)

Run the fitting process.

Parameters:

Name	Type	Description	Default
sample_models	SampleModels	Collection of sample models.	required
experiments	Experiments	Collection of experiments.	required
calculator	Any	The calculator to use for pattern generation.	required
weights	Optional[array]	Optional weights for joint fitting.	None

Source code in src/easydiffraction/analysis/fitting.py

def fit(
    self,
    sample_models: SampleModels,
    experiments: Experiments,
    calculator: Any,
    weights: Optional[np.array] = None,
) -> None:
    """Run the fitting process.

    Args:
        sample_models: Collection of sample models.
        experiments: Collection of experiments.
        calculator: The calculator to use for pattern generation.
        weights: Optional weights for joint fitting.
    """
    params = sample_models.free_parameters + experiments.free_parameters

    if not params:
        print('⚠️ No parameters selected for fitting.')
        return None

    for param in params:
        param._fit_start_value = param.value

    def objective_function(engine_params: Dict[str, Any]) -> np.ndarray:
        return self._residual_function(
            engine_params=engine_params,
            parameters=params,
            sample_models=sample_models,
            experiments=experiments,
            calculator=calculator,
            weights=weights,
        )

    # Perform fitting
    self.results = self.minimizer.fit(params, objective_function)

    # Post-fit processing
    self._process_fit_results(sample_models, experiments, calculator)

minimizers

base

MinimizerBase

Bases: ABC

Abstract base for concrete minimizers.

Contract: - Subclasses must implement _prepare_solver_args, _run_solver, _sync_result_to_parameters and _check_success. - The fit method orchestrates the full workflow and returns :class:FitResults.

Source code in src/easydiffraction/analysis/minimizers/base.py

class MinimizerBase(ABC):
    """Abstract base for concrete minimizers.

    Contract:
    - Subclasses must implement ``_prepare_solver_args``,
        ``_run_solver``, ``_sync_result_to_parameters`` and
        ``_check_success``.
    - The ``fit`` method orchestrates the full workflow and returns
        :class:`FitResults`.
    """

    def __init__(
        self,
        name: Optional[str] = None,
        method: Optional[str] = None,
        max_iterations: Optional[int] = None,
    ) -> None:
        self.name: Optional[str] = name
        self.method: Optional[str] = method
        self.max_iterations: Optional[int] = max_iterations
        self.result: Optional[FitResults] = None
        self._previous_chi2: Optional[float] = None
        self._iteration: Optional[int] = None
        self._best_chi2: Optional[float] = None
        self._best_iteration: Optional[int] = None
        self._fitting_time: Optional[float] = None
        self.tracker: FitProgressTracker = FitProgressTracker()

    def _start_tracking(self, minimizer_name: str) -> None:
        """Initialize progress tracking and timer.

        Args:
            minimizer_name: Human-readable name shown in progress.
        """
        self.tracker.reset()
        self.tracker.start_tracking(minimizer_name)
        self.tracker.start_timer()

    def _stop_tracking(self) -> None:
        """Stop timer and finalize tracking."""
        self.tracker.stop_timer()
        self.tracker.finish_tracking()

    @abstractmethod
    def _prepare_solver_args(self, parameters: List[Any]) -> Dict[str, Any]:
        """Prepare keyword-arguments for the underlying solver.

        Args:
            parameters: List of free parameters to be fitted.

        Returns:
            Mapping of keyword arguments to pass into ``_run_solver``.
        """
        pass

    @abstractmethod
    def _run_solver(
        self,
        objective_function: Callable[..., Any],
        engine_parameters: Dict[str, Any],
    ) -> Any:
        """Execute the concrete solver and return its raw result."""
        pass

    @abstractmethod
    def _sync_result_to_parameters(
        self,
        raw_result: Any,
        parameters: List[Any],
    ) -> None:
        """Copy values from ``raw_result`` back to ``parameters`` in-
        place.
        """
        pass

    def _finalize_fit(
        self,
        parameters: List[Any],
        raw_result: Any,
    ) -> FitResults:
        """Build :class:`FitResults` and store it on ``self.result``.

        Args:
            parameters: Parameters after the solver finished.
            raw_result: Backend-specific solver output object.

        Returns:
            FitResults: Aggregated outcome of the fit.
        """
        self._sync_result_to_parameters(parameters, raw_result)
        success = self._check_success(raw_result)
        self.result = FitResults(
            success=success,
            parameters=parameters,
            reduced_chi_square=self.tracker.best_chi2,
            engine_result=raw_result,
            starting_parameters=parameters,
            fitting_time=self.tracker.fitting_time,
        )
        return self.result

    @abstractmethod
    def _check_success(self, raw_result: Any) -> bool:
        """Determine whether the fit was successful."""
        pass

    def fit(
        self,
        parameters: List[Any],
        objective_function: Callable[..., Any],
    ) -> FitResults:
        """Run the full minimization workflow.

        Args:
            parameters: Free parameters to optimize.
            objective_function: Callable returning residuals for a given
                set of engine arguments.

        Returns:
            FitResults with success flag, best chi2 and timing.
        """
        minimizer_name = self.name or 'Unnamed Minimizer'
        if self.method is not None:
            minimizer_name += f' ({self.method})'

        self._start_tracking(minimizer_name)

        solver_args = self._prepare_solver_args(parameters)
        raw_result = self._run_solver(objective_function, **solver_args)

        self._stop_tracking()

        result = self._finalize_fit(parameters, raw_result)

        return result

    def _objective_function(
        self,
        engine_params: Dict[str, Any],
        parameters: List[Any],
        sample_models: Any,
        experiments: Any,
        calculator: Any,
    ) -> np.ndarray:
        """Default objective helper computing residuals array."""
        return self._compute_residuals(
            engine_params,
            parameters,
            sample_models,
            experiments,
            calculator,
        )

    def _create_objective_function(
        self,
        parameters: List[Any],
        sample_models: Any,
        experiments: Any,
        calculator: Any,
    ) -> Callable[[Dict[str, Any]], np.ndarray]:
        """Return a closure capturing problem context for the solver."""
        return lambda engine_params: self._objective_function(
            engine_params,
            parameters,
            sample_models,
            experiments,
            calculator,
        )

fit(parameters, objective_function)

Run the full minimization workflow.

Parameters:

Name	Type	Description	Default
parameters	List[Any]	Free parameters to optimize.	required
objective_function	Callable[..., Any]	Callable returning residuals for a given set of engine arguments.	required

Returns:

Type	Description
FitResults	FitResults with success flag, best chi2 and timing.

Source code in src/easydiffraction/analysis/minimizers/base.py

def fit(
    self,
    parameters: List[Any],
    objective_function: Callable[..., Any],
) -> FitResults:
    """Run the full minimization workflow.

    Args:
        parameters: Free parameters to optimize.
        objective_function: Callable returning residuals for a given
            set of engine arguments.

    Returns:
        FitResults with success flag, best chi2 and timing.
    """
    minimizer_name = self.name or 'Unnamed Minimizer'
    if self.method is not None:
        minimizer_name += f' ({self.method})'

    self._start_tracking(minimizer_name)

    solver_args = self._prepare_solver_args(parameters)
    raw_result = self._run_solver(objective_function, **solver_args)

    self._stop_tracking()

    result = self._finalize_fit(parameters, raw_result)

    return result

dfols

DfolsMinimizer

Bases: MinimizerBase

Minimizer using the DFO-LS package (Derivative-Free Optimization for Least-Squares).

Source code in src/easydiffraction/analysis/minimizers/dfols.py

class DfolsMinimizer(MinimizerBase):
    """Minimizer using the DFO-LS package (Derivative-Free Optimization
    for Least-Squares).
    """

    def __init__(
        self,
        name: str = 'dfols',
        max_iterations: int = DEFAULT_MAX_ITERATIONS,
        **kwargs: Any,
    ) -> None:
        super().__init__(name=name, method=None, max_iterations=max_iterations)
        # Intentionally unused, accepted for API compatibility
        del kwargs

    def _prepare_solver_args(self, parameters: List[Any]) -> Dict[str, Any]:
        x0 = []
        bounds_lower = []
        bounds_upper = []
        for param in parameters:
            x0.append(param.value)
            bounds_lower.append(param.fit_min)
            bounds_upper.append(param.fit_max)
        bounds = (np.array(bounds_lower), np.array(bounds_upper))
        return {'x0': np.array(x0), 'bounds': bounds}

    def _run_solver(self, objective_function: Any, **kwargs: Any) -> Any:
        x0 = kwargs.get('x0')
        bounds = kwargs.get('bounds')
        return solve(objective_function, x0=x0, bounds=bounds, maxfun=self.max_iterations)

    def _sync_result_to_parameters(
        self,
        parameters: List[Any],
        raw_result: Any,
    ) -> None:
        """Synchronizes the result from the solver to the parameters.

        Args:
            parameters: List of parameters being optimized.
            raw_result: The result object returned by the solver.
        """
        # Ensure compatibility with raw_result coming from dfols.solve()
        result_values = raw_result.x if hasattr(raw_result, 'x') else raw_result

        for i, param in enumerate(parameters):
            param.value = result_values[i]
            # DFO-LS doesn't provide uncertainties; set to None or
            # calculate later if needed
            param.uncertainty = None

    def _check_success(self, raw_result: Any) -> bool:
        """Determines success from DFO-LS result dictionary.

        Args:
            raw_result: The result object returned by the solver.

        Returns:
            True if the optimization was successful, False otherwise.
        """
        return raw_result.flag == raw_result.EXIT_SUCCESS

factory

MinimizerFactory

Source code in src/easydiffraction/analysis/minimizers/factory.py

class MinimizerFactory:
    _available_minimizers: Dict[str, Dict[str, Any]] = {
        'lmfit': {
            'engine': 'lmfit',
            'method': 'leastsq',
            'description': 'LMFIT library using the default Levenberg-Marquardt '
            'least squares method',
            'class': LmfitMinimizer,
        },
        'lmfit (leastsq)': {
            'engine': 'lmfit',
            'method': 'leastsq',
            'description': 'LMFIT library with Levenberg-Marquardt least squares method',
            'class': LmfitMinimizer,
        },
        'lmfit (least_squares)': {
            'engine': 'lmfit',
            'method': 'least_squares',
            'description': 'LMFIT library with SciPy’s trust region reflective algorithm',
            'class': LmfitMinimizer,
        },
        'dfols': {
            'engine': 'dfols',
            'method': None,
            'description': 'DFO-LS library for derivative-free least-squares optimization',
            'class': DfolsMinimizer,
        },
    }

    @classmethod
    def list_available_minimizers(cls) -> List[str]:
        """List all available minimizers.

        Returns:
            A list of minimizer names.
        """
        return list(cls._available_minimizers.keys())

    @classmethod
    def show_available_minimizers(cls) -> None:
        # TODO: Rename this method to `show_supported_minimizers` for
        #  consistency with other methods in the library. E.g.
        #  `show_supported_calculators`, etc.
        """Display a table of available minimizers and their
        descriptions.
        """
        columns_headers: List[str] = ['Minimizer', 'Description']
        columns_alignment = ['left', 'left']
        columns_data: List[List[str]] = []
        for name, config in cls._available_minimizers.items():
            description: str = config.get('description', 'No description provided.')
            columns_data.append([name, description])

        console.paragraph('Supported minimizers')
        render_table(
            columns_headers=columns_headers,
            columns_alignment=columns_alignment,
            columns_data=columns_data,
        )

    @classmethod
    def create_minimizer(cls, selection: str) -> MinimizerBase:
        """Create a minimizer instance based on the selection.

        Args:
            selection: The name of the minimizer to create.

        Returns:
            An instance of the selected minimizer.

        Raises:
            ValueError: If the selection is not a valid minimizer.
        """
        config = cls._available_minimizers.get(selection)
        if not config:
            raise ValueError(
                f"Unknown minimizer '{selection}'. Use one of {cls.list_available_minimizers()}"
            )

        minimizer_class: Type[MinimizerBase] = config.get('class')
        method: Optional[str] = config.get('method')

        kwargs: Dict[str, Any] = {}
        if method is not None:
            kwargs['method'] = method

        return minimizer_class(**kwargs)

    @classmethod
    def register_minimizer(
        cls,
        name: str,
        minimizer_cls: Type[MinimizerBase],
        method: Optional[str] = None,
        description: str = 'No description provided.',
    ) -> None:
        """Register a new minimizer.

        Args:
            name: The name of the minimizer.
            minimizer_cls: The class of the minimizer.
            method: The method used by the minimizer (optional).
            description: A description of the minimizer.
        """
        cls._available_minimizers[name] = {
            'engine': name,
            'method': method,
            'description': description,
            'class': minimizer_cls,
        }

create_minimizer(selection) classmethod

Create a minimizer instance based on the selection.

Parameters:

Name	Type	Description	Default
selection	str	The name of the minimizer to create.	required

Returns:

Type	Description
MinimizerBase	An instance of the selected minimizer.

Raises:

Type	Description
ValueError	If the selection is not a valid minimizer.

Source code in src/easydiffraction/analysis/minimizers/factory.py

@classmethod
def create_minimizer(cls, selection: str) -> MinimizerBase:
    """Create a minimizer instance based on the selection.

    Args:
        selection: The name of the minimizer to create.

    Returns:
        An instance of the selected minimizer.

    Raises:
        ValueError: If the selection is not a valid minimizer.
    """
    config = cls._available_minimizers.get(selection)
    if not config:
        raise ValueError(
            f"Unknown minimizer '{selection}'. Use one of {cls.list_available_minimizers()}"
        )

    minimizer_class: Type[MinimizerBase] = config.get('class')
    method: Optional[str] = config.get('method')

    kwargs: Dict[str, Any] = {}
    if method is not None:
        kwargs['method'] = method

    return minimizer_class(**kwargs)

list_available_minimizers() classmethod

List all available minimizers.

Returns:

Type	Description
List[str]	A list of minimizer names.

Source code in src/easydiffraction/analysis/minimizers/factory.py

@classmethod
def list_available_minimizers(cls) -> List[str]:
    """List all available minimizers.

    Returns:
        A list of minimizer names.
    """
    return list(cls._available_minimizers.keys())

register_minimizer(name, minimizer_cls, method=None, description='No description provided.') classmethod

Register a new minimizer.

Parameters:

Name	Type	Description	Default
name	str	The name of the minimizer.	required
minimizer_cls	Type[MinimizerBase]	The class of the minimizer.	required
method	Optional[str]	The method used by the minimizer (optional).	None
description	str	A description of the minimizer.	'No description provided.'

Source code in src/easydiffraction/analysis/minimizers/factory.py

@classmethod
def register_minimizer(
    cls,
    name: str,
    minimizer_cls: Type[MinimizerBase],
    method: Optional[str] = None,
    description: str = 'No description provided.',
) -> None:
    """Register a new minimizer.

    Args:
        name: The name of the minimizer.
        minimizer_cls: The class of the minimizer.
        method: The method used by the minimizer (optional).
        description: A description of the minimizer.
    """
    cls._available_minimizers[name] = {
        'engine': name,
        'method': method,
        'description': description,
        'class': minimizer_cls,
    }

show_available_minimizers() classmethod

Display a table of available minimizers and their descriptions.

Source code in src/easydiffraction/analysis/minimizers/factory.py

@classmethod
def show_available_minimizers(cls) -> None:
    # TODO: Rename this method to `show_supported_minimizers` for
    #  consistency with other methods in the library. E.g.
    #  `show_supported_calculators`, etc.
    """Display a table of available minimizers and their
    descriptions.
    """
    columns_headers: List[str] = ['Minimizer', 'Description']
    columns_alignment = ['left', 'left']
    columns_data: List[List[str]] = []
    for name, config in cls._available_minimizers.items():
        description: str = config.get('description', 'No description provided.')
        columns_data.append([name, description])

    console.paragraph('Supported minimizers')
    render_table(
        columns_headers=columns_headers,
        columns_alignment=columns_alignment,
        columns_data=columns_data,
    )

lmfit

LmfitMinimizer

Bases: MinimizerBase

Minimizer using the lmfit package.

Source code in src/easydiffraction/analysis/minimizers/lmfit.py

class LmfitMinimizer(MinimizerBase):
    """Minimizer using the lmfit package."""

    def __init__(
        self,
        name: str = 'lmfit',
        method: str = DEFAULT_METHOD,
        max_iterations: int = DEFAULT_MAX_ITERATIONS,
    ) -> None:
        super().__init__(
            name=name,
            method=method,
            max_iterations=max_iterations,
        )

    def _prepare_solver_args(
        self,
        parameters: List[Any],
    ) -> Dict[str, Any]:
        """Prepares the solver arguments for the lmfit minimizer.

        Args:
            parameters: List of parameters to be optimized.

        Returns:
            A dictionary containing the prepared lmfit. Parameters
                object.
        """
        engine_parameters = lmfit.Parameters()
        for param in parameters:
            engine_parameters.add(
                name=param._minimizer_uid,
                value=param.value,
                vary=param.free,
                min=param.fit_min,
                max=param.fit_max,
            )
        return {'engine_parameters': engine_parameters}

    def _run_solver(self, objective_function: Any, **kwargs: Any) -> Any:
        """Runs the lmfit solver.

        Args:
            objective_function: The objective function to minimize.
            **kwargs: Additional arguments for the solver.

        Returns:
            The result of the lmfit minimization.
        """
        engine_parameters = kwargs.get('engine_parameters')

        return lmfit.minimize(
            objective_function,
            params=engine_parameters,
            method=self.method,
            nan_policy='propagate',
            max_nfev=self.max_iterations,
        )

    def _sync_result_to_parameters(
        self,
        parameters: List[Any],
        raw_result: Any,
    ) -> None:
        """Synchronizes the result from the solver to the parameters.

        Args:
            parameters: List of parameters being optimized.
            raw_result: The result object returned by the solver.
        """
        param_values = raw_result.params if hasattr(raw_result, 'params') else raw_result

        for param in parameters:
            param_result = param_values.get(param._minimizer_uid)
            if param_result is not None:
                param._value = param_result.value  # Bypass ranges check
                param.uncertainty = getattr(param_result, 'stderr', None)

    def _check_success(self, raw_result: Any) -> bool:
        """Determines success from lmfit MinimizerResult.

        Args:
            raw_result: The result object returned by the solver.

        Returns:
            True if the optimization was successful, False otherwise.
        """
        return getattr(raw_result, 'success', False)

    def _iteration_callback(
        self,
        params: lmfit.Parameters,
        iter: int,
        resid: Any,
        *args: Any,
        **kwargs: Any,
    ) -> None:
        """Callback function for each iteration of the minimizer.

        Args:
            params: The current parameters.
            iter: The current iteration number.
            resid: The residuals.
            *args: Additional positional arguments.
            **kwargs: Additional keyword arguments.
        """
        # Intentionally unused, required by callback signature
        del params, resid, args, kwargs
        self._iteration = iter