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 structures and experiments.

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 structures and experiments.
    """

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

        Parameters
        ----------
        project : object
            The project that owns models and experiments.
        """
        self.project = project
        self._aliases_type: str = AliasesFactory.default_tag()
        self.aliases = AliasesFactory.create(self._aliases_type)
        self._constraints_type: str = ConstraintsFactory.default_tag()
        self.constraints = ConstraintsFactory.create(self._constraints_type)
        self.constraints_handler = ConstraintsHandler.get()
        self._fit_mode_type: str = FitModeFactory.default_tag()
        self._fit_mode = FitModeFactory.create(self._fit_mode_type)
        self._joint_fit_experiments = JointFitExperiments()
        self.fitter = Fitter('lmfit')
        self.fit_results = None
        self._parameter_snapshots: dict[str, dict[str, dict]] = {}

    def help(self) -> None:
        """Print a summary of analysis properties and methods."""
        from easydiffraction.core.guard import GuardedBase

        console.paragraph("Help for 'Analysis'")

        cls = type(self)

        # Auto-discover properties from MRO
        seen_props: dict = {}
        for base in cls.mro():
            for key, attr in base.__dict__.items():
                if key.startswith('_') or not isinstance(attr, property):
                    continue
                if key not in seen_props:
                    seen_props[key] = attr

        prop_rows = []
        for i, key in enumerate(sorted(seen_props), 1):
            prop = seen_props[key]
            writable = '✓' if prop.fset else '✗'
            doc = GuardedBase._first_sentence(prop.fget.__doc__ if prop.fget else None)
            prop_rows.append([str(i), key, writable, doc])

        if prop_rows:
            console.paragraph('Properties')
            render_table(
                columns_headers=['#', 'Name', 'Writable', 'Description'],
                columns_alignment=['right', 'left', 'center', 'left'],
                columns_data=prop_rows,
            )

        # Auto-discover methods from MRO
        seen_methods: set = set()
        methods_list: list = []
        for base in cls.mro():
            for key, attr in base.__dict__.items():
                if key.startswith('_') or key in seen_methods:
                    continue
                if isinstance(attr, property):
                    continue
                raw = attr
                if isinstance(raw, (staticmethod, classmethod)):
                    raw = raw.__func__
                if callable(raw):
                    seen_methods.add(key)
                    methods_list.append((key, raw))

        method_rows = []
        for i, (key, method) in enumerate(sorted(methods_list), 1):
            doc = GuardedBase._first_sentence(getattr(method, '__doc__', None))
            method_rows.append([str(i), f'{key}()', doc])

        if method_rows:
            console.paragraph('Methods')
            render_table(
                columns_headers=['#', 'Name', 'Description'],
                columns_alignment=['right', 'left', 'left'],
                columns_data=method_rows,
            )

    # ------------------------------------------------------------------
    #  Aliases (switchable-category pattern)
    # ------------------------------------------------------------------

    @property
    def aliases_type(self) -> str:
        """Tag of the active aliases collection type."""
        return self._aliases_type

    @aliases_type.setter
    def aliases_type(self, new_type: str) -> None:
        """
        Switch to a different aliases collection type.

        Parameters
        ----------
        new_type : str
            Aliases tag (e.g. ``'default'``).
        """
        supported_tags = AliasesFactory.supported_tags()
        if new_type not in supported_tags:
            log.warning(
                f"Unsupported aliases type '{new_type}'. "
                f'Supported: {supported_tags}. '
                f"For more information, use 'show_supported_aliases_types()'",
            )
            return
        self.aliases = AliasesFactory.create(new_type)
        self._aliases_type = new_type
        console.paragraph('Aliases type changed to')
        console.print(new_type)

    def show_supported_aliases_types(self) -> None:
        """Print a table of supported aliases collection types."""
        AliasesFactory.show_supported()

    def show_current_aliases_type(self) -> None:
        """Print the currently used aliases collection type."""
        console.paragraph('Current aliases type')
        console.print(self._aliases_type)

    # ------------------------------------------------------------------
    #  Constraints (switchable-category pattern)
    # ------------------------------------------------------------------

    @property
    def constraints_type(self) -> str:
        """Tag of the active constraints collection type."""
        return self._constraints_type

    @constraints_type.setter
    def constraints_type(self, new_type: str) -> None:
        """
        Switch to a different constraints collection type.

        Parameters
        ----------
        new_type : str
            Constraints tag (e.g. ``'default'``).
        """
        supported_tags = ConstraintsFactory.supported_tags()
        if new_type not in supported_tags:
            log.warning(
                f"Unsupported constraints type '{new_type}'. "
                f'Supported: {supported_tags}. '
                f"For more information, use 'show_supported_constraints_types()'",
            )
            return
        self.constraints = ConstraintsFactory.create(new_type)
        self._constraints_type = new_type
        console.paragraph('Constraints type changed to')
        console.print(new_type)

    def show_supported_constraints_types(self) -> None:
        """Print a table of supported constraints collection types."""
        ConstraintsFactory.show_supported()

    def show_current_constraints_type(self) -> None:
        """Print the currently used constraints collection type."""
        console.paragraph('Current constraints type')
        console.print(self._constraints_type)

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

        Parameters
        ----------
        params : List[Union[NumericDescriptor, Parameter]]
            List of DescriptorFloat or Parameter objects.

        Returns
        -------
        pd.DataFrame
            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 all parameters for structures and experiments."""
        structures_params = self.project.structures.parameters
        experiments_params = self.project.experiments.parameters

        if not structures_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 structures (🧩 data blocks)')
        df = self._get_params_as_dataframe(structures_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 all fittable parameters."""
        structures_params = self.project.structures.fittable_parameters
        experiments_params = self.project.experiments.fittable_parameters

        if not structures_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 structures (🧩 data blocks)')
        df = self._get_params_as_dataframe(structures_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 only currently free (varying) parameters."""
        structures_params = self.project.structures.free_parameters
        experiments_params = self.project.experiments.free_parameters
        free_params = structures_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 structures (🧩 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.
        """
        structures_params = self.project.structures.parameters
        experiments_params = self.project.experiments.parameters
        all_params = {
            'structures': structures_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.
        """
        structures_params = self.project.structures.parameters
        experiments_params = self.project.experiments.parameters
        all_params = {
            'structures': structures_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_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 available minimizer drivers on this system."""
        MinimizerFactory.show_supported()

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

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

    # ------------------------------------------------------------------
    #  Fit mode (switchable-category pattern)
    # ------------------------------------------------------------------

    @property
    def fit_mode(self) -> object:
        """Fit-mode category item holding the active strategy."""
        return self._fit_mode

    @property
    def fit_mode_type(self) -> str:
        """Tag of the active fit-mode category type."""
        return self._fit_mode_type

    @fit_mode_type.setter
    def fit_mode_type(self, new_type: str) -> None:
        """
        Switch to a different fit-mode category type.

        Parameters
        ----------
        new_type : str
            Fit-mode tag (e.g. ``'default'``).
        """
        supported_tags = FitModeFactory.supported_tags()
        if new_type not in supported_tags:
            log.warning(
                f"Unsupported fit-mode type '{new_type}'. "
                f'Supported: {supported_tags}. '
                f"For more information, use 'show_supported_fit_mode_types()'",
            )
            return
        self._fit_mode = FitModeFactory.create(new_type)
        self._fit_mode_type = new_type
        console.paragraph('Fit-mode type changed to')
        console.print(new_type)

    def show_supported_fit_mode_types(self) -> None:
        """Print a table of supported fit-mode category types."""
        FitModeFactory.show_supported()

    def show_current_fit_mode_type(self) -> None:
        """Print the currently used fit-mode category type."""
        console.paragraph('Current fit-mode type')
        console.print(self._fit_mode_type)

    # ------------------------------------------------------------------
    #  Joint-fit experiments (category)
    # ------------------------------------------------------------------

    @property
    def joint_fit_experiments(self) -> object:
        """Per-experiment weight collection for joint fitting."""
        return self._joint_fit_experiments

    def show_constraints(self) -> None:
        """Print a table of all user-defined symbolic constraints."""
        if not self.constraints._items:
            log.warning('No constraints defined.')
            return

        rows = []
        for constraint in self.constraints:
            rows.append([constraint.expression.value])

        console.paragraph('User defined constraints')
        render_table(
            columns_headers=['expression'],
            columns_alignment=['left'],
            columns_data=rows,
        )

    def apply_constraints(self) -> None:
        """Apply currently defined constraints to the 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, verbosity: str | None = None) -> None:
        """
        Execute fitting for all experiments.

        This method performs the optimization but does not display
        results automatically. Call :meth:`show_fit_results` after
        fitting to see a summary of the fit quality and parameter
        values.

        In 'single' mode, fits each experiment independently. In 'joint'
        mode, performs a simultaneous fit across experiments with
        weights.

        Sets :attr:`fit_results` on success, which can be accessed
        programmatically (e.g.,
        ``analysis.fit_results.reduced_chi_square``).

        Parameters
        ----------
        verbosity : str | None, default=None
            Console output verbosity: ``'full'`` for detailed per-
            experiment progress, ``'short'`` for a
            one-row-per-experiment summary table, or ``'silent'`` for no
            output. When ``None``, uses ``project.verbosity``.

        Raises
        ------
        NotImplementedError
            If the fit mode is not ``'single'`` or ``'joint'``.
        """
        verb = VerbosityEnum(verbosity if verbosity is not None else self.project.verbosity)

        structures = self.project.structures
        if not structures:
            log.warning('No structures 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

        # Run the fitting process
        mode = FitModeEnum(self._fit_mode.mode.value)
        if mode is FitModeEnum.JOINT:
            # Auto-populate joint_fit_experiments if empty
            if not len(self._joint_fit_experiments):
                for id in experiments.names:
                    self._joint_fit_experiments.create(id=id, weight=0.5)
            if verb is not VerbosityEnum.SILENT:
                console.paragraph(
                    f"Using all experiments 🔬 {experiments.names} for '{mode.value}' fitting"
                )
            self.fitter.fit(
                structures,
                experiments,
                weights=self._joint_fit_experiments,
                analysis=self,
                verbosity=verb,
            )

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

        elif mode is FitModeEnum.SINGLE:
            expt_names = experiments.names
            num_expts = len(expt_names)

            # Short mode: print header and create display handle once
            short_headers = ['experiment', 'χ²', 'iterations', 'status']
            short_alignments = ['left', 'right', 'right', 'center']
            short_rows: list[list[str]] = []
            short_display_handle: object | None = None
            if verb is VerbosityEnum.SHORT:
                from easydiffraction.analysis.fit_helpers.tracking import _make_display_handle

                first = expt_names[0]
                last = expt_names[-1]
                minimizer_name = self.fitter.selection
                console.paragraph(
                    f"Using {num_expts} experiments 🔬 from '{first}' to "
                    f"'{last}' for '{mode.value}' fitting"
                )
                console.print(f"🚀 Starting fit process with '{minimizer_name}'...")
                console.print('📈 Goodness-of-fit (reduced χ²) per experiment:')
                short_display_handle = _make_display_handle()

            # TODO: Find a better way without creating dummy
            #  experiments?
            for _idx, expt_name in enumerate(expt_names, start=1):
                if verb is VerbosityEnum.FULL:
                    console.paragraph(
                        f"Using experiment 🔬 '{expt_name}' for '{mode.value}' fitting"
                    )

                experiment = experiments[expt_name]
                dummy_experiments = Experiments()  # TODO: Find a better name

                # This is a workaround to set the parent project
                # of the dummy experiments collection, so that
                # parameters can be resolved correctly during fitting.
                object.__setattr__(dummy_experiments, '_parent', self.project)

                dummy_experiments.add(experiment)
                self.fitter.fit(
                    structures,
                    dummy_experiments,
                    analysis=self,
                    verbosity=verb,
                )

                # After fitting, snapshot parameter values before
                # they get overwritten by the next experiment's fit
                results = self.fitter.results
                snapshot: dict[str, dict] = {}
                for param in results.parameters:
                    snapshot[param.unique_name] = {
                        'value': param.value,
                        'uncertainty': param.uncertainty,
                        'units': param.units,
                    }
                self._parameter_snapshots[expt_name] = snapshot
                self.fit_results = results

                # Short mode: append one summary row and update in-place
                if verb is VerbosityEnum.SHORT:
                    chi2_str = (
                        f'{results.reduced_chi_square:.2f}'
                        if results.reduced_chi_square is not None
                        else '—'
                    )
                    iters = str(self.fitter.minimizer.tracker.best_iteration or 0)
                    status = '✅' if results.success else '❌'
                    short_rows.append([expt_name, chi2_str, iters, status])
                    render_table(
                        columns_headers=short_headers,
                        columns_alignment=short_alignments,
                        columns_data=short_rows,
                        display_handle=short_display_handle,
                    )

            # Short mode: close the display handle
            if short_display_handle is not None and hasattr(short_display_handle, 'close'):
                from contextlib import suppress

                with suppress(Exception):
                    short_display_handle.close()

        else:
            raise NotImplementedError(f'Fit mode {mode.value} not implemented yet.')

        # After fitting, save the project
        # TODO: Consider saving individual data during sequential
        #  (single) fitting, instead of waiting until the end and save
        #  only the last one
        if self.project.info.path is not None:
            self.project.save()

    def show_fit_results(self) -> None:
        """
        Display a summary of the fit results.

        Renders the fit quality metrics (reduced χ², R-factors) and a
        table of fitted parameters with their starting values, final
        values, and uncertainties.

        This method should be called after :meth:`fit` completes. If no
        fit has been performed yet, a warning is logged.

        Example::

        project.analysis.fit() project.analysis.show_fit_results()
        """
        if self.fit_results is None:
            log.warning('No fit results available. Run fit() first.')
            return

        structures = self.project.structures
        experiments = self.project.experiments

        self.fitter._process_fit_results(structures, experiments)

    def _update_categories(self, called_by_minimizer: bool = False) -> None:
        """
        Update all categories owned by Analysis.

        This ensures aliases and constraints are up-to-date before
        serialization or after parameter changes.

        Parameters
        ----------
        called_by_minimizer : bool, default=False
            Whether this is called during fitting.
        """
        # Apply constraints to sync dependent parameters
        if self.constraints._items:
            self.constraints_handler.apply()

        # Update category-specific logic
        # TODO: Need self.categories as in the case of datablock.py
        for category in [self.aliases, self.constraints]:
            if hasattr(category, '_update'):
                category._update(called_by_minimizer=called_by_minimizer)

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

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

        self._update_categories()
        return analysis_to_cif(self)

    def show_as_cif(self) -> None:
        """Render the analysis section as CIF in console."""
        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	object	The project that owns models and experiments.	required

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

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

    Parameters
    ----------
    project : object
        The project that owns models and experiments.
    """
    self.project = project
    self._aliases_type: str = AliasesFactory.default_tag()
    self.aliases = AliasesFactory.create(self._aliases_type)
    self._constraints_type: str = ConstraintsFactory.default_tag()
    self.constraints = ConstraintsFactory.create(self._constraints_type)
    self.constraints_handler = ConstraintsHandler.get()
    self._fit_mode_type: str = FitModeFactory.default_tag()
    self._fit_mode = FitModeFactory.create(self._fit_mode_type)
    self._joint_fit_experiments = JointFitExperiments()
    self.fitter = Fitter('lmfit')
    self.fit_results = None
    self._parameter_snapshots: dict[str, dict[str, dict]] = {}

aliases_type property writable

Tag of the active aliases collection type.

apply_constraints()

Apply currently defined constraints to the project.

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

def apply_constraints(self) -> None:
    """Apply currently defined constraints to the 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
str	The analysis section represented as a CIF document string.

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

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

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

    self._update_categories()
    return analysis_to_cif(self)

constraints_type property writable

Tag of the active constraints collection type.

current_minimizer property writable

The identifier of the active minimizer, if any.

fit(verbosity=None)

Execute fitting for all experiments.

This method performs the optimization but does not display results automatically. Call :meth:show_fit_results after fitting to see a summary of the fit quality and parameter values.

In 'single' mode, fits each experiment independently. In 'joint' mode, performs a simultaneous fit across experiments with weights.

Sets :attr:fit_results on success, which can be accessed programmatically (e.g., analysis.fit_results.reduced_chi_square).

Parameters:

Name	Type	Description	Default
verbosity	str | None	Console output verbosity: 'full' for detailed per- experiment progress, 'short' for a one-row-per-experiment summary table, or 'silent' for no output. When None, uses project.verbosity.	None

Raises:

Type	Description
NotImplementedError	If the fit mode is not 'single' or 'joint'.

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

def fit(self, verbosity: str | None = None) -> None:
    """
    Execute fitting for all experiments.

    This method performs the optimization but does not display
    results automatically. Call :meth:`show_fit_results` after
    fitting to see a summary of the fit quality and parameter
    values.

    In 'single' mode, fits each experiment independently. In 'joint'
    mode, performs a simultaneous fit across experiments with
    weights.

    Sets :attr:`fit_results` on success, which can be accessed
    programmatically (e.g.,
    ``analysis.fit_results.reduced_chi_square``).

    Parameters
    ----------
    verbosity : str | None, default=None
        Console output verbosity: ``'full'`` for detailed per-
        experiment progress, ``'short'`` for a
        one-row-per-experiment summary table, or ``'silent'`` for no
        output. When ``None``, uses ``project.verbosity``.

    Raises
    ------
    NotImplementedError
        If the fit mode is not ``'single'`` or ``'joint'``.
    """
    verb = VerbosityEnum(verbosity if verbosity is not None else self.project.verbosity)

    structures = self.project.structures
    if not structures:
        log.warning('No structures 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

    # Run the fitting process
    mode = FitModeEnum(self._fit_mode.mode.value)
    if mode is FitModeEnum.JOINT:
        # Auto-populate joint_fit_experiments if empty
        if not len(self._joint_fit_experiments):
            for id in experiments.names:
                self._joint_fit_experiments.create(id=id, weight=0.5)
        if verb is not VerbosityEnum.SILENT:
            console.paragraph(
                f"Using all experiments 🔬 {experiments.names} for '{mode.value}' fitting"
            )
        self.fitter.fit(
            structures,
            experiments,
            weights=self._joint_fit_experiments,
            analysis=self,
            verbosity=verb,
        )

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

    elif mode is FitModeEnum.SINGLE:
        expt_names = experiments.names
        num_expts = len(expt_names)

        # Short mode: print header and create display handle once
        short_headers = ['experiment', 'χ²', 'iterations', 'status']
        short_alignments = ['left', 'right', 'right', 'center']
        short_rows: list[list[str]] = []
        short_display_handle: object | None = None
        if verb is VerbosityEnum.SHORT:
            from easydiffraction.analysis.fit_helpers.tracking import _make_display_handle

            first = expt_names[0]
            last = expt_names[-1]
            minimizer_name = self.fitter.selection
            console.paragraph(
                f"Using {num_expts} experiments 🔬 from '{first}' to "
                f"'{last}' for '{mode.value}' fitting"
            )
            console.print(f"🚀 Starting fit process with '{minimizer_name}'...")
            console.print('📈 Goodness-of-fit (reduced χ²) per experiment:')
            short_display_handle = _make_display_handle()

        # TODO: Find a better way without creating dummy
        #  experiments?
        for _idx, expt_name in enumerate(expt_names, start=1):
            if verb is VerbosityEnum.FULL:
                console.paragraph(
                    f"Using experiment 🔬 '{expt_name}' for '{mode.value}' fitting"
                )

            experiment = experiments[expt_name]
            dummy_experiments = Experiments()  # TODO: Find a better name

            # This is a workaround to set the parent project
            # of the dummy experiments collection, so that
            # parameters can be resolved correctly during fitting.
            object.__setattr__(dummy_experiments, '_parent', self.project)

            dummy_experiments.add(experiment)
            self.fitter.fit(
                structures,
                dummy_experiments,
                analysis=self,
                verbosity=verb,
            )

            # After fitting, snapshot parameter values before
            # they get overwritten by the next experiment's fit
            results = self.fitter.results
            snapshot: dict[str, dict] = {}
            for param in results.parameters:
                snapshot[param.unique_name] = {
                    'value': param.value,
                    'uncertainty': param.uncertainty,
                    'units': param.units,
                }
            self._parameter_snapshots[expt_name] = snapshot
            self.fit_results = results

            # Short mode: append one summary row and update in-place
            if verb is VerbosityEnum.SHORT:
                chi2_str = (
                    f'{results.reduced_chi_square:.2f}'
                    if results.reduced_chi_square is not None
                    else '—'
                )
                iters = str(self.fitter.minimizer.tracker.best_iteration or 0)
                status = '✅' if results.success else '❌'
                short_rows.append([expt_name, chi2_str, iters, status])
                render_table(
                    columns_headers=short_headers,
                    columns_alignment=short_alignments,
                    columns_data=short_rows,
                    display_handle=short_display_handle,
                )

        # Short mode: close the display handle
        if short_display_handle is not None and hasattr(short_display_handle, 'close'):
            from contextlib import suppress

            with suppress(Exception):
                short_display_handle.close()

    else:
        raise NotImplementedError(f'Fit mode {mode.value} not implemented yet.')

    # After fitting, save the project
    # TODO: Consider saving individual data during sequential
    #  (single) fitting, instead of waiting until the end and save
    #  only the last one
    if self.project.info.path is not None:
        self.project.save()

fit_mode property

Fit-mode category item holding the active strategy.

fit_mode_type property writable

Tag of the active fit-mode category type.

help()

Print a summary of analysis properties and methods.

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

def help(self) -> None:
    """Print a summary of analysis properties and methods."""
    from easydiffraction.core.guard import GuardedBase

    console.paragraph("Help for 'Analysis'")

    cls = type(self)

    # Auto-discover properties from MRO
    seen_props: dict = {}
    for base in cls.mro():
        for key, attr in base.__dict__.items():
            if key.startswith('_') or not isinstance(attr, property):
                continue
            if key not in seen_props:
                seen_props[key] = attr

    prop_rows = []
    for i, key in enumerate(sorted(seen_props), 1):
        prop = seen_props[key]
        writable = '✓' if prop.fset else '✗'
        doc = GuardedBase._first_sentence(prop.fget.__doc__ if prop.fget else None)
        prop_rows.append([str(i), key, writable, doc])

    if prop_rows:
        console.paragraph('Properties')
        render_table(
            columns_headers=['#', 'Name', 'Writable', 'Description'],
            columns_alignment=['right', 'left', 'center', 'left'],
            columns_data=prop_rows,
        )

    # Auto-discover methods from MRO
    seen_methods: set = set()
    methods_list: list = []
    for base in cls.mro():
        for key, attr in base.__dict__.items():
            if key.startswith('_') or key in seen_methods:
                continue
            if isinstance(attr, property):
                continue
            raw = attr
            if isinstance(raw, (staticmethod, classmethod)):
                raw = raw.__func__
            if callable(raw):
                seen_methods.add(key)
                methods_list.append((key, raw))

    method_rows = []
    for i, (key, method) in enumerate(sorted(methods_list), 1):
        doc = GuardedBase._first_sentence(getattr(method, '__doc__', None))
        method_rows.append([str(i), f'{key}()', doc])

    if method_rows:
        console.paragraph('Methods')
        render_table(
            columns_headers=['#', 'Name', 'Description'],
            columns_alignment=['right', 'left', 'left'],
            columns_data=method_rows,
        )

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.
    """
    structures_params = self.project.structures.parameters
    experiments_params = self.project.experiments.parameters
    all_params = {
        'structures': structures_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,
    )

joint_fit_experiments property

Per-experiment weight collection for joint fitting.

show_all_params()

Print all parameters for structures and experiments.

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

def show_all_params(self) -> None:
    """Print all parameters for structures and experiments."""
    structures_params = self.project.structures.parameters
    experiments_params = self.project.experiments.parameters

    if not structures_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 structures (🧩 data blocks)')
    df = self._get_params_as_dataframe(structures_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 console.

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

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

show_available_minimizers() staticmethod

Print available minimizer drivers on this system.

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

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

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."""
    if not self.constraints._items:
        log.warning('No constraints defined.')
        return

    rows = []
    for constraint in self.constraints:
        rows.append([constraint.expression.value])

    console.paragraph('User defined constraints')
    render_table(
        columns_headers=['expression'],
        columns_alignment=['left'],
        columns_data=rows,
    )

show_current_aliases_type()

Print the currently used aliases collection type.

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

def show_current_aliases_type(self) -> None:
    """Print the currently used aliases collection type."""
    console.paragraph('Current aliases type')
    console.print(self._aliases_type)

show_current_constraints_type()

Print the currently used constraints collection type.

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

def show_current_constraints_type(self) -> None:
    """Print the currently used constraints collection type."""
    console.paragraph('Current constraints type')
    console.print(self._constraints_type)

show_current_fit_mode_type()

Print the currently used fit-mode category type.

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

def show_current_fit_mode_type(self) -> None:
    """Print the currently used fit-mode category type."""
    console.paragraph('Current fit-mode type')
    console.print(self._fit_mode_type)

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_fit_results()

Display a summary of the fit results.

Renders the fit quality metrics (reduced χ², R-factors) and a table of fitted parameters with their starting values, final values, and uncertainties.

This method should be called after :meth:fit completes. If no fit has been performed yet, a warning is logged.

Example::

project.analysis.fit() project.analysis.show_fit_results()

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

def show_fit_results(self) -> None:
    """
    Display a summary of the fit results.

    Renders the fit quality metrics (reduced χ², R-factors) and a
    table of fitted parameters with their starting values, final
    values, and uncertainties.

    This method should be called after :meth:`fit` completes. If no
    fit has been performed yet, a warning is logged.

    Example::

    project.analysis.fit() project.analysis.show_fit_results()
    """
    if self.fit_results is None:
        log.warning('No fit results available. Run fit() first.')
        return

    structures = self.project.structures
    experiments = self.project.experiments

    self.fitter._process_fit_results(structures, experiments)

show_fittable_params()

Print all fittable parameters.

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

def show_fittable_params(self) -> None:
    """Print all fittable parameters."""
    structures_params = self.project.structures.fittable_parameters
    experiments_params = self.project.experiments.fittable_parameters

    if not structures_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 structures (🧩 data blocks)')
    df = self._get_params_as_dataframe(structures_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 only currently free (varying) parameters.

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

def show_free_params(self) -> None:
    """Print only currently free (varying) parameters."""
    structures_params = self.project.structures.free_parameters
    experiments_params = self.project.experiments.free_parameters
    free_params = structures_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 structures (🧩 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.
    """
    structures_params = self.project.structures.parameters
    experiments_params = self.project.experiments.parameters
    all_params = {
        'structures': structures_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_aliases_types()

Print a table of supported aliases collection types.

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

def show_supported_aliases_types(self) -> None:
    """Print a table of supported aliases collection types."""
    AliasesFactory.show_supported()

show_supported_constraints_types()

Print a table of supported constraints collection types.

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

def show_supported_constraints_types(self) -> None:
    """Print a table of supported constraints collection types."""
    ConstraintsFactory.show_supported()

show_supported_fit_mode_types()

Print a table of supported fit-mode category types.

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

def show_supported_fit_mode_types(self) -> None:
    """Print a table of supported fit-mode category types."""
    FitModeFactory.show_supported()

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:
        """Short identifier of the calculation engine."""
        pass

    @property
    @abstractmethod
    def engine_imported(self) -> bool:
        """True if the underlying calculation library is available."""
        pass

    @abstractmethod
    def calculate_structure_factors(
        self,
        structure: Structure,
        experiment: ExperimentBase,
        called_by_minimizer: bool,
    ) -> None:
        """Calculate structure factors for one experiment."""
        pass

    @abstractmethod
    def calculate_pattern(
        self,
        structure: Structures,  # TODO: Structure?
        experiment: ExperimentBase,
        called_by_minimizer: bool,
    ) -> np.ndarray:
        """
        Calculate diffraction pattern for one structure-experiment pair.

        Parameters
        ----------
        structure : Structures
            The structure object.
        experiment : ExperimentBase
            The experiment object.
        called_by_minimizer : bool
            Whether the calculation is called by a minimizer. Default is
            False.

        Returns
        -------
        np.ndarray
            The calculated diffraction pattern as a NumPy array.
        """
        pass

calculate_pattern(structure, experiment, called_by_minimizer) abstractmethod

Calculate diffraction pattern for one structure-experiment pair.

Parameters:

Name	Type	Description	Default
structure	Structures	The structure object.	required
experiment	ExperimentBase	The experiment object.	required
called_by_minimizer	bool	Whether the calculation is called by a minimizer. Default is False.	required

Returns:

Type	Description
ndarray	The calculated diffraction pattern as a NumPy array.

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

@abstractmethod
def calculate_pattern(
    self,
    structure: Structures,  # TODO: Structure?
    experiment: ExperimentBase,
    called_by_minimizer: bool,
) -> np.ndarray:
    """
    Calculate diffraction pattern for one structure-experiment pair.

    Parameters
    ----------
    structure : Structures
        The structure object.
    experiment : ExperimentBase
        The experiment object.
    called_by_minimizer : bool
        Whether the calculation is called by a minimizer. Default is
        False.

    Returns
    -------
    np.ndarray
        The calculated diffraction pattern as a NumPy array.
    """
    pass

calculate_structure_factors(structure, experiment, called_by_minimizer) abstractmethod

Calculate structure factors for one experiment.

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

@abstractmethod
def calculate_structure_factors(
    self,
    structure: Structure,
    experiment: ExperimentBase,
    called_by_minimizer: bool,
) -> None:
    """Calculate structure factors for one experiment."""
    pass

engine_imported abstractmethod property

True if the underlying calculation library is available.

name abstractmethod property

Short identifier of the calculation engine.

crysfml

CrysfmlCalculator

Bases: CalculatorBase

Wrapper for Crysfml library.

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

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

    type_info = TypeInfo(
        tag='crysfml',
        description='CrysFML library for crystallographic calculations',
    )
    engine_imported: bool = cfml_py_utilities is not None

    @property
    def name(self) -> str:
        """Short identifier of this calculator engine."""
        return 'crysfml'

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

        Parameters
        ----------
        structures : Structures
            The structures to calculate structure factors for.
        experiments : Experiments
            The experiments associated with the sample models.

        Raises
        ------
        NotImplementedError
            HKL calculation is not implemented for CrysfmlCalculator.
        """
        raise NotImplementedError('HKL calculation is not implemented for CrysfmlCalculator.')

    def calculate_pattern(
        self,
        structure: Structures,
        experiment: ExperimentBase,
        called_by_minimizer: bool = False,
    ) -> Union[np.ndarray, List[float]]:
        """
        Calculate the diffraction pattern using Crysfml.

        Parameters
        ----------
        structure : Structures
            The structure to calculate the pattern for.
        experiment : ExperimentBase
            The experiment associated with the structure.
        called_by_minimizer : bool, default=False
            Whether the calculation is called by a minimizer.

        Returns
        -------
        Union[np.ndarray, List[float]]
            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(structure, experiment)
        try:
            _, y = cfml_py_utilities.cw_powder_pattern_from_dict(crysfml_dict)
            y = self._adjust_pattern_length(y, len(experiment.data.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]:
        """
        Adjust the pattern length to match the target length.

        Parameters
        ----------
        pattern : List[float]
            The pattern to adjust.
        target_length : int
            The desired length of the pattern.

        Returns
        -------
        List[float]
            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,
        structure: Structures,
        experiment: ExperimentBase,
    ) -> Dict[str, Union[ExperimentBase, Structure]]:
        """
        Convert structure and experiment into a Crysfml dictionary.

        Parameters
        ----------
        structure : Structures
            The structure to convert.
        experiment : ExperimentBase
            The experiment to convert.

        Returns
        -------
        Dict[str, Union[ExperimentBase, Structure]]
            A dictionary representation of the structure and experiment.
        """
        structure_dict = self._convert_structure_to_dict(structure)
        experiment_dict = self._convert_experiment_to_dict(experiment)
        return {
            'phases': [structure_dict],
            'experiments': [experiment_dict],
        }

    def _convert_structure_to_dict(
        self,
        structure: Structure,
    ) -> Dict[str, Any]:
        """
        Convert a structure into a dictionary format.

        Parameters
        ----------
        structure : Structure
            The structure to convert.

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

        for atom in structure.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,
            }
            structure_dict[structure.name]['_atom_site'].append(atom_site)

        return structure_dict

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

        Parameters
        ----------
        experiment : ExperimentBase
            The experiment to convert.

        Returns
        -------
        Dict[str, Any]
            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.data.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_pattern(structure, experiment, called_by_minimizer=False)

Calculate the diffraction pattern using Crysfml.

Parameters:

Name	Type	Description	Default
structure	Structures	The structure to calculate the pattern for.	required
experiment	ExperimentBase	The experiment associated with the structure.	required
called_by_minimizer	bool	Whether the calculation is called by a minimizer.	False

Returns:

Type	Description
Union[ndarray, List[float]]	The calculated diffraction pattern as a NumPy array or a list of floats.

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

def calculate_pattern(
    self,
    structure: Structures,
    experiment: ExperimentBase,
    called_by_minimizer: bool = False,
) -> Union[np.ndarray, List[float]]:
    """
    Calculate the diffraction pattern using Crysfml.

    Parameters
    ----------
    structure : Structures
        The structure to calculate the pattern for.
    experiment : ExperimentBase
        The experiment associated with the structure.
    called_by_minimizer : bool, default=False
        Whether the calculation is called by a minimizer.

    Returns
    -------
    Union[np.ndarray, List[float]]
        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(structure, experiment)
    try:
        _, y = cfml_py_utilities.cw_powder_pattern_from_dict(crysfml_dict)
        y = self._adjust_pattern_length(y, len(experiment.data.x))
    except KeyError:
        print('[CrysfmlCalculator] Error: No calculated data')
        y = []
    return y

calculate_structure_factors(structures, experiments)

Call Crysfml to calculate structure factors.

Parameters:

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

Raises:

Type	Description
NotImplementedError	HKL calculation is not implemented for CrysfmlCalculator.

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

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

    Parameters
    ----------
    structures : Structures
        The structures to calculate structure factors for.
    experiments : Experiments
        The experiments associated with the sample models.

    Raises
    ------
    NotImplementedError
        HKL calculation is not implemented for CrysfmlCalculator.
    """
    raise NotImplementedError('HKL calculation is not implemented for CrysfmlCalculator.')

name property

Short identifier of this calculator engine.

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

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

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

    type_info = TypeInfo(
        tag='cryspy',
        description='CrysPy library for crystallographic calculations',
    )
    engine_imported: bool = cryspy is not None

    @property
    def name(self) -> str:
        """Short identifier of this calculator engine."""
        return 'cryspy'

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

    def calculate_structure_factors(
        self,
        structure: Structure,
        experiment: ExperimentBase,
        called_by_minimizer: bool = False,
    ) -> None:
        """
        Raise NotImplementedError as HKL calculation is not implemented.

        Parameters
        ----------
        structure : Structure
            The structure to calculate structure factors for.
        experiment : ExperimentBase
            The experiment associated with the sample models.
        called_by_minimizer : bool, default=False
            Whether the calculation is called by a minimizer.
        """
        combined_name = f'{structure.name}_{experiment.name}'

        if called_by_minimizer:
            if self._cryspy_dicts and combined_name in self._cryspy_dicts:
                cryspy_dict = self._recreate_cryspy_dict(structure, experiment)
            else:
                cryspy_obj = self._recreate_cryspy_obj(structure, experiment)
                cryspy_dict = cryspy_obj.get_dictionary()
        else:
            cryspy_obj = self._recreate_cryspy_obj(structure, 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,
            )

        cryspy_block_name = f'diffrn_{experiment.name}'

        try:
            y_calc = cryspy_in_out_dict[cryspy_block_name]['intensity_calc']
            stol = cryspy_in_out_dict[cryspy_block_name]['sthovl']
        except KeyError:
            print(f'[CryspyCalculator] Error: No calculated data for {cryspy_block_name}')
            return [], []

        return stol, y_calc

    def calculate_pattern(
        self,
        structure: Structure,
        experiment: ExperimentBase,
        called_by_minimizer: bool = False,
    ) -> Union[np.ndarray, List[float]]:
        """
        Calculate the diffraction pattern using Cryspy.

        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

        Parameters
        ----------
        structure : Structure
            The structure to calculate the pattern for.
        experiment : ExperimentBase
            The experiment associated with the structure.
        called_by_minimizer : bool, default=False
            Whether the calculation is called by a minimizer.

        Returns
        -------
        Union[np.ndarray, List[float]]
            The calculated diffraction pattern as a NumPy array or a
            list of floats.
        """
        combined_name = f'{structure.name}_{experiment.name}'

        if called_by_minimizer:
            if self._cryspy_dicts and combined_name in self._cryspy_dicts:
                cryspy_dict = self._recreate_cryspy_dict(structure, experiment)
            else:
                cryspy_obj = self._recreate_cryspy_obj(structure, experiment)
                cryspy_dict = cryspy_obj.get_dictionary()
        else:
            cryspy_obj = self._recreate_cryspy_obj(structure, 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,
        structure: Structure,
        experiment: ExperimentBase,
    ) -> Dict[str, Any]:
        """
        Recreate the Cryspy dictionary for structure and experiment.

        Parameters
        ----------
        structure : Structure
            The structure to update.
        experiment : ExperimentBase
            The experiment to update.

        Returns
        -------
        Dict[str, Any]
            The updated Cryspy dictionary.
        """
        combined_name = f'{structure.name}_{experiment.name}'
        cryspy_dict = copy.deepcopy(self._cryspy_dicts[combined_name])

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

        ################################
        # Update structure parameters
        ################################

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

        # Atomic coordinates
        cryspy_xyz = cryspy_model_dict['atom_fract_xyz']
        for idx, atom_site in enumerate(structure.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(structure.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(structure.atom_sites):
            cryspy_biso[idx] = atom_site.b_iso.value

        ##############################
        # Update experiment parameters
        ##############################

        if experiment.type.sample_form.value == SampleFormEnum.POWDER:
            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

        if experiment.type.sample_form.value == SampleFormEnum.SINGLE_CRYSTAL:
            cryspy_expt_name = f'diffrn_{experiment.name}'
            cryspy_expt_dict = cryspy_dict[cryspy_expt_name]

            # Instrument
            if experiment.type.beam_mode.value == BeamModeEnum.CONSTANT_WAVELENGTH:
                cryspy_expt_dict['wavelength'][0] = experiment.instrument.setup_wavelength.value

            # Extinction
            cryspy_expt_dict['extinction_radius'][0] = experiment.extinction.radius.value
            cryspy_expt_dict['extinction_mosaicity'][0] = experiment.extinction.mosaicity.value

        return cryspy_dict

    def _recreate_cryspy_obj(
        self,
        structure: Structure,
        experiment: ExperimentBase,
    ) -> object:
        """
        Recreate the Cryspy object for structure and experiment.

        Parameters
        ----------
        structure : Structure
            The structure to recreate.
        experiment : ExperimentBase
            The experiment to recreate.

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

        cryspy_structure_cif = self._convert_structure_to_cryspy_cif(structure)
        cryspy_structure_obj = str_to_globaln(cryspy_structure_cif)
        cryspy_obj.add_items(cryspy_structure_obj.items)

        # Add single experiment to cryspy_obj
        cryspy_experiment_cif = self._convert_experiment_to_cryspy_cif(
            experiment,
            linked_structure=structure,
        )

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

        return cryspy_obj

    def _convert_structure_to_cryspy_cif(
        self,
        structure: Structure,
    ) -> str:
        """
        Convert a structure to a Cryspy CIF string.

        Parameters
        ----------
        structure : Structure
            The structure to convert.

        Returns
        -------
        str
            The Cryspy CIF string representation of the structure.
        """
        return structure.as_cif

    def _convert_experiment_to_cryspy_cif(
        self,
        experiment: ExperimentBase,
        linked_structure: object,
    ) -> str:
        """
        Convert an experiment to a Cryspy CIF string.

        Parameters
        ----------
        experiment : ExperimentBase
            The experiment to convert.
        linked_structure : object
            The structure linked to the experiment.

        Returns
        -------
        str
            The Cryspy CIF string representation of the experiment.
        """
        # Try to get experiment attributes
        expt_type = getattr(experiment, 'type', None)
        instrument = getattr(experiment, 'instrument', None)
        peak = getattr(experiment, 'peak', None)
        extinction = getattr(experiment, 'extinction', None)

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

        # Add experiment type attribute dat
        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}')

        # Add instrument attribute data
        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:
                if expt_type.sample_form.value == SampleFormEnum.POWDER:
                    instrument_mapping = {
                        'setup_wavelength': '_setup_wavelength',
                        'calib_twotheta_offset': '_setup_offset_2theta',
                    }
                elif expt_type.sample_form.value == SampleFormEnum.SINGLE_CRYSTAL:
                    instrument_mapping = {
                        'setup_wavelength': '_setup_wavelength',
                    }
                    # Add dummy 0.0 value for _setup_field required by
                    # Cryspy
                    cif_lines.append('')
                    cif_lines.append('_setup_field 0.0')
            elif expt_type.beam_mode.value == BeamModeEnum.TIME_OF_FLIGHT:
                if expt_type.sample_form.value == SampleFormEnum.POWDER:
                    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',
                    }
                elif expt_type.sample_form.value == SampleFormEnum.SINGLE_CRYSTAL:
                    instrument_mapping = {}  # TODO: Check this mapping!
                    # Add dummy 0.0 value for _setup_field required by
                    # Cryspy
                    cif_lines.append('')
                    cif_lines.append('_setup_field 0.0')
            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}')

        # Add peak attribute data
        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}')

        # Add extinction attribute data
        if extinction and expt_type.sample_form.value == SampleFormEnum.SINGLE_CRYSTAL:
            extinction_mapping = {
                'mosaicity': '_extinction_mosaicity',
                'radius': '_extinction_radius',
            }
            cif_lines.append('')
            cif_lines.append('_extinction_model gauss')
            for local_attr_name, engine_key_name in extinction_mapping.items():
                attr_obj = getattr(extinction, local_attr_name)
                if attr_obj is not None:
                    cif_lines.append(f'{engine_key_name} {attr_obj.value}')

        # Add range data
        if expt_type.sample_form.value == SampleFormEnum.POWDER:
            x_data = experiment.data.x
            twotheta_min = f'{np.round(x_data.min(), 5):.5f}'  # float(x_data.min())
            twotheta_max = f'{np.round(x_data.max(), 5):.5f}'  # 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}')

        # Add orientation matrix data
        # Hardcoded example values for now, as we don't use them yet,
        # but Cryspy requires them for single crystal data.
        if expt_type.sample_form.value == SampleFormEnum.SINGLE_CRYSTAL:
            cif_lines.append('')
            cif_lines.append('_diffrn_orient_matrix_type CCSL')
            cif_lines.append('_diffrn_orient_matrix_ub_11 -0.088033')
            cif_lines.append('_diffrn_orient_matrix_ub_12 -0.088004')
            cif_lines.append('_diffrn_orient_matrix_ub_13  0.069970')
            cif_lines.append('_diffrn_orient_matrix_ub_21  0.034058')
            cif_lines.append('_diffrn_orient_matrix_ub_22 -0.188170')
            cif_lines.append('_diffrn_orient_matrix_ub_23 -0.013039')
            cif_lines.append('_diffrn_orient_matrix_ub_31  0.223600')
            cif_lines.append('_diffrn_orient_matrix_ub_32  0.125751')
            cif_lines.append('_diffrn_orient_matrix_ub_33  0.029490')

        # Add phase data
        if expt_type.sample_form.value == SampleFormEnum.SINGLE_CRYSTAL:
            cif_lines.append('')
            cif_lines.append(f'_phase_label {linked_structure.name}')
            cif_lines.append('_phase_scale 1.0')
        elif expt_type.sample_form.value == SampleFormEnum.POWDER:
            cif_lines.append('')
            cif_lines.append('loop_')
            cif_lines.append('_phase_label')
            cif_lines.append('_phase_scale')
            cif_lines.append(f'{linked_structure.name} 1.0')

        # Add background data
        if expt_type.sample_form.value == SampleFormEnum.POWDER:
            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')  # TODO: !!!!????
                cif_lines.append('_tof_backgroundpoint_intensity')  # TODO: !!!!????
                cif_lines.append(f'{twotheta_min} 0.0')  # TODO: !!!!????
                cif_lines.append(f'{twotheta_max} 0.0')  # TODO: !!!!????

        # Add measured data: Single crystal
        if expt_type.sample_form.value == SampleFormEnum.SINGLE_CRYSTAL:
            if expt_type.beam_mode.value == BeamModeEnum.CONSTANT_WAVELENGTH:
                cif_lines.append('')
                cif_lines.append('loop_')
                cif_lines.append('_diffrn_refln_index_h')
                cif_lines.append('_diffrn_refln_index_k')
                cif_lines.append('_diffrn_refln_index_l')
                cif_lines.append('_diffrn_refln_intensity')
                cif_lines.append('_diffrn_refln_intensity_sigma')
                indices_h: np.ndarray = experiment.data.index_h
                indices_k: np.ndarray = experiment.data.index_k
                indices_l: np.ndarray = experiment.data.index_l
                y_data: np.ndarray = experiment.data.intensity_meas
                sy_data: np.ndarray = experiment.data.intensity_meas_su
                for index_h, index_k, index_l, y_val, sy_val in zip(
                    indices_h, indices_k, indices_l, y_data, sy_data, strict=True
                ):
                    cif_lines.append(
                        f'{index_h:4.0f}{index_k:4.0f}{index_l:4.0f}   {y_val:.5f}   {sy_val:.5f}'
                    )
            elif expt_type.beam_mode.value == BeamModeEnum.TIME_OF_FLIGHT:
                cif_lines.append('')
                cif_lines.append('loop_')
                cif_lines.append('_diffrn_refln_index_h')
                cif_lines.append('_diffrn_refln_index_k')
                cif_lines.append('_diffrn_refln_index_l')
                cif_lines.append('_diffrn_refln_intensity')
                cif_lines.append('_diffrn_refln_intensity_sigma')
                cif_lines.append('_diffrn_refln_wavelength')
                indices_h: np.ndarray = experiment.data.index_h
                indices_k: np.ndarray = experiment.data.index_k
                indices_l: np.ndarray = experiment.data.index_l
                y_data: np.ndarray = experiment.data.intensity_meas
                sy_data: np.ndarray = experiment.data.intensity_meas_su
                wl_data: np.ndarray = experiment.data.wavelength
                for index_h, index_k, index_l, y_val, sy_val, wl_val in zip(
                    indices_h, indices_k, indices_l, y_data, sy_data, wl_data, strict=True
                ):
                    cif_lines.append(
                        f'{index_h:4.0f}{index_k:4.0f}{index_l:4.0f}   {y_val:.5f}   '
                        f'{sy_val:.5f}   {wl_val:.5f}'
                    )
        # Add measured data: Powder
        elif expt_type.sample_form.value == SampleFormEnum.POWDER:
            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: np.ndarray = experiment.data.intensity_meas
            sy_data: np.ndarray = experiment.data.intensity_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}')

        # Combine all lines into a single CIF string
        cryspy_experiment_cif = '\n'.join(cif_lines)

        return cryspy_experiment_cif

calculate_pattern(structure, experiment, called_by_minimizer=False)

Calculate the diffraction pattern using Cryspy.

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

Parameters:

Name	Type	Description	Default
structure	Structure	The structure to calculate the pattern for.	required
experiment	ExperimentBase	The experiment associated with the structure.	required
called_by_minimizer	bool	Whether the calculation is called by a minimizer.	False

Returns:

Type	Description
Union[ndarray, List[float]]	The calculated diffraction pattern as a NumPy array or a list of floats.

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

def calculate_pattern(
    self,
    structure: Structure,
    experiment: ExperimentBase,
    called_by_minimizer: bool = False,
) -> Union[np.ndarray, List[float]]:
    """
    Calculate the diffraction pattern using Cryspy.

    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

    Parameters
    ----------
    structure : Structure
        The structure to calculate the pattern for.
    experiment : ExperimentBase
        The experiment associated with the structure.
    called_by_minimizer : bool, default=False
        Whether the calculation is called by a minimizer.

    Returns
    -------
    Union[np.ndarray, List[float]]
        The calculated diffraction pattern as a NumPy array or a
        list of floats.
    """
    combined_name = f'{structure.name}_{experiment.name}'

    if called_by_minimizer:
        if self._cryspy_dicts and combined_name in self._cryspy_dicts:
            cryspy_dict = self._recreate_cryspy_dict(structure, experiment)
        else:
            cryspy_obj = self._recreate_cryspy_obj(structure, experiment)
            cryspy_dict = cryspy_obj.get_dictionary()
    else:
        cryspy_obj = self._recreate_cryspy_obj(structure, 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

calculate_structure_factors(structure, experiment, called_by_minimizer=False)

Raise NotImplementedError as HKL calculation is not implemented.

Parameters:

Name	Type	Description	Default
structure	Structure	The structure to calculate structure factors for.	required
experiment	ExperimentBase	The experiment associated with the sample models.	required
called_by_minimizer	bool	Whether the calculation is called by a minimizer.	False

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

def calculate_structure_factors(
    self,
    structure: Structure,
    experiment: ExperimentBase,
    called_by_minimizer: bool = False,
) -> None:
    """
    Raise NotImplementedError as HKL calculation is not implemented.

    Parameters
    ----------
    structure : Structure
        The structure to calculate structure factors for.
    experiment : ExperimentBase
        The experiment associated with the sample models.
    called_by_minimizer : bool, default=False
        Whether the calculation is called by a minimizer.
    """
    combined_name = f'{structure.name}_{experiment.name}'

    if called_by_minimizer:
        if self._cryspy_dicts and combined_name in self._cryspy_dicts:
            cryspy_dict = self._recreate_cryspy_dict(structure, experiment)
        else:
            cryspy_obj = self._recreate_cryspy_obj(structure, experiment)
            cryspy_dict = cryspy_obj.get_dictionary()
    else:
        cryspy_obj = self._recreate_cryspy_obj(structure, 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,
        )

    cryspy_block_name = f'diffrn_{experiment.name}'

    try:
        y_calc = cryspy_in_out_dict[cryspy_block_name]['intensity_calc']
        stol = cryspy_in_out_dict[cryspy_block_name]['sthovl']
    except KeyError:
        print(f'[CryspyCalculator] Error: No calculated data for {cryspy_block_name}')
        return [], []

    return stol, y_calc

name property

Short identifier of this calculator engine.

factory

Calculator factory — delegates to FactoryBase.

Overrides _supported_map to filter out calculators whose engines are not importable in the current environment.

CalculatorFactory

Bases: FactoryBase

Factory for creating calculation engine instances.

Only calculators whose engine_imported flag is True are available for creation.

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

class CalculatorFactory(FactoryBase):
    """
    Factory for creating calculation engine instances.

    Only calculators whose ``engine_imported`` flag is ``True`` are
    available for creation.
    """

    _default_rules = {
        frozenset({
            ('scattering_type', ScatteringTypeEnum.BRAGG),
        }): CalculatorEnum.CRYSPY,
        frozenset({
            ('scattering_type', ScatteringTypeEnum.TOTAL),
        }): CalculatorEnum.PDFFIT,
    }

    @classmethod
    def _supported_map(cls) -> Dict[str, Type]:
        """Only include calculators whose engines are importable."""
        return {klass.type_info.tag: klass for klass in cls._registry if klass.engine_imported}

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

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

    type_info = TypeInfo(
        tag='pdffit',
        description='PDFfit2 for pair distribution function calculations',
    )
    engine_imported: bool = PdfFit is not None

    @property
    def name(self) -> str:
        """Short identifier of this calculator engine."""
        return 'pdffit'

    def calculate_structure_factors(
        self,
        structures: object,
        experiments: object,
    ) -> list:
        """
        Return an empty list; PDF does not compute structure factors.

        Parameters
        ----------
        structures : object
            Unused; kept for interface consistency.
        experiments : object
            Unused; kept for interface consistency.

        Returns
        -------
        list
            An empty list.
        """
        # PDF doesn't compute HKL but we keep interface consistent
        # Intentionally unused, required by public API/signature
        del structures, experiments
        print('[pdffit] Calculating HKLs (not applicable)...')
        return []

    def calculate_pattern(
        self,
        structure: Structure,
        experiment: ExperimentBase,
        called_by_minimizer: bool = False,
    ) -> None:
        """
        Calculate the PDF pattern using PDFfit2.

        Parameters
        ----------
        structure : Structure
            The structure object supplying atom sites and cell
            parameters.
        experiment : ExperimentBase
            The experiment object supplying instrument and peak
            parameters.
        called_by_minimizer : bool, default=False
            Unused; kept for interface consistency.
        """
        # Intentionally unused, required by public API/signature
        del called_by_minimizer

        # Create PDF calculator object
        calculator = PdfFit()

        # ---------------------------
        # Set structure parameters
        # ---------------------------

        # TODO: move CIF v2 -> CIF v1 conversion to a separate module
        # Convert the structure to CIF supported by PDFfit
        cif_string_v2 = structure.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
        pdffit_structure = pdffit_cif_parser().parse(cif_string_v1)

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

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

        # Set some peak-related parameters
        calculator.setvar('pscale', experiment.linked_phases[structure.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.data.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

calculate_pattern(structure, experiment, called_by_minimizer=False)

Calculate the PDF pattern using PDFfit2.

Parameters:

Name	Type	Description	Default
structure	Structure	The structure object supplying atom sites and cell parameters.	required
experiment	ExperimentBase	The experiment object supplying instrument and peak parameters.	required
called_by_minimizer	bool	Unused; kept for interface consistency.	False

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

def calculate_pattern(
    self,
    structure: Structure,
    experiment: ExperimentBase,
    called_by_minimizer: bool = False,
) -> None:
    """
    Calculate the PDF pattern using PDFfit2.

    Parameters
    ----------
    structure : Structure
        The structure object supplying atom sites and cell
        parameters.
    experiment : ExperimentBase
        The experiment object supplying instrument and peak
        parameters.
    called_by_minimizer : bool, default=False
        Unused; kept for interface consistency.
    """
    # Intentionally unused, required by public API/signature
    del called_by_minimizer

    # Create PDF calculator object
    calculator = PdfFit()

    # ---------------------------
    # Set structure parameters
    # ---------------------------

    # TODO: move CIF v2 -> CIF v1 conversion to a separate module
    # Convert the structure to CIF supported by PDFfit
    cif_string_v2 = structure.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
    pdffit_structure = pdffit_cif_parser().parse(cif_string_v1)

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

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

    # Set some peak-related parameters
    calculator.setvar('pscale', experiment.linked_phases[structure.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.data.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

calculate_structure_factors(structures, experiments)

Return an empty list; PDF does not compute structure factors.

Parameters:

Name	Type	Description	Default
structures	object	Unused; kept for interface consistency.	required
experiments	object	Unused; kept for interface consistency.	required

Returns:

Type	Description
list	An empty list.

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

def calculate_structure_factors(
    self,
    structures: object,
    experiments: object,
) -> list:
    """
    Return an empty list; PDF does not compute structure factors.

    Parameters
    ----------
    structures : object
        Unused; kept for interface consistency.
    experiments : object
        Unused; kept for interface consistency.

    Returns
    -------
    list
        An empty list.
    """
    # PDF doesn't compute HKL but we keep interface consistent
    # Intentionally unused, required by public API/signature
    del structures, experiments
    print('[pdffit] Calculating HKLs (not applicable)...')
    return []

name property

Short identifier of this calculator engine.

categories

aliases

default

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.

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

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

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

    def __init__(self) -> None:
        super().__init__()

        self._label = StringDescriptor(
            name='label',
            description='...',  # TODO
            value_spec=AttributeSpec(
                default='_',  # TODO, Maybe None?
                validator=RegexValidator(pattern=r'^[A-Za-z_][A-Za-z0-9_]*$'),
            ),
            cif_handler=CifHandler(names=['_alias.label']),
        )
        self._param_uid = StringDescriptor(
            name='param_uid',
            description='...',  # TODO
            value_spec=AttributeSpec(
                default='_',
                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)

    # ------------------------------------------------------------------
    #  Public properties
    # ------------------------------------------------------------------

    @property
    def label(self) -> StringDescriptor:
        """
        ...

        Reading this property returns the underlying
        ``StringDescriptor`` object. Assigning to it updates the
        parameter value.
        """
        return self._label

    @label.setter
    def label(self, value: str) -> None:
        self._label.value = value

    @property
    def param_uid(self) -> StringDescriptor:
        """
        ...

        Reading this property returns the underlying
        ``StringDescriptor`` object. Assigning to it updates the
        parameter value.
        """
        return self._param_uid

    @param_uid.setter
    def param_uid(self, value: str) -> None:
        self._param_uid.value = value

label property writable

...

Reading this property returns the underlying StringDescriptor object. Assigning to it updates the parameter value.

param_uid property writable

...

Reading this property returns the underlying StringDescriptor object. Assigning to it updates the parameter value.

Aliases

Bases: CategoryCollection

Collection of :class:Alias items.

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

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

    type_info = TypeInfo(
        tag='default',
        description='Parameter alias mappings',
    )

    def __init__(self) -> None:
        """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/default.py

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

factory

Aliases factory — delegates entirely to FactoryBase.

AliasesFactory

Bases: FactoryBase

Create alias collections by tag.

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

class AliasesFactory(FactoryBase):
    """Create alias collections by tag."""

    _default_rules = {
        frozenset(): 'default',
    }

constraints

default

Simple symbolic constraint between parameters.

Represents an equation of the form lhs_alias = rhs_expr stored as a single expression string. The left- and right-hand sides are derived by splitting the expression at the = sign.

Constraint

Bases: CategoryItem

Single constraint item stored as lhs = rhs expression.

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

class Constraint(CategoryItem):
    """Single constraint item stored as ``lhs = rhs`` expression."""

    def __init__(self) -> None:
        super().__init__()

        self._expression = StringDescriptor(
            name='expression',
            description='Constraint equation, e.g. "occ_Ba = 1 - occ_La".',
            value_spec=AttributeSpec(
                default='_',  # TODO, Maybe None?
                validator=RegexValidator(pattern=r'.*'),
            ),
            cif_handler=CifHandler(names=['_constraint.expression']),
        )

        self._identity.category_code = 'constraint'
        self._identity.category_entry_name = lambda: self.lhs_alias

    # ------------------------------------------------------------------
    #  Public properties
    # ------------------------------------------------------------------

    @property
    def expression(self) -> StringDescriptor:
        """
        Full constraint equation (e.g. ``'occ_Ba = 1 - occ_La'``).

        Reading this property returns the underlying
        ``StringDescriptor`` object. Assigning to it updates the value.
        """
        return self._expression

    @expression.setter
    def expression(self, value: str) -> None:
        self._expression.value = value

    @property
    def lhs_alias(self) -> str:
        """Left-hand side alias derived from the expression."""
        return self._split_expression()[0]

    @property
    def rhs_expr(self) -> str:
        """Right-hand side expression derived from the expression."""
        return self._split_expression()[1]

    # ------------------------------------------------------------------
    #  Internal helpers
    # ------------------------------------------------------------------

    def _split_expression(self) -> tuple[str, str]:
        """
        Split the expression at the first ``=`` sign.

        Returns
        -------
        tuple[str, str]
            ``(lhs_alias, rhs_expr)`` with whitespace stripped.
        """
        raw = self._expression.value or ''
        if '=' not in raw:
            return (raw.strip(), '')
        lhs, rhs = raw.split('=', 1)
        return (lhs.strip(), rhs.strip())

expression property writable

Full constraint equation (e.g. 'occ_Ba = 1 - occ_La').

Reading this property returns the underlying StringDescriptor object. Assigning to it updates the value.

lhs_alias property

Left-hand side alias derived from the expression.

rhs_expr property

Right-hand side expression derived from the expression.

Constraints

Bases: CategoryCollection

Collection of :class:Constraint items.

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

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

    type_info = TypeInfo(
        tag='default',
        description='Symbolic parameter constraints',
    )

    _update_priority = 90  # After most others, but before data categories

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

    def create(self, *, expression: str) -> None:
        """
        Create a constraint from an expression string.

        Parameters
        ----------
        expression : str
            Constraint equation, e.g. ``'biso_Co2 = biso_Co1'`` or
            ``'occ_Ba = 1 - occ_La'``.
        """
        item = Constraint()
        item.expression = expression
        self.add(item)

    def _update(self, called_by_minimizer: bool = False) -> None:
        del called_by_minimizer

        constraints = ConstraintsHandler.get()
        constraints.apply()

__init__()

Create an empty constraints collection.

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

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

create(*, expression)

Create a constraint from an expression string.

Parameters:

Name	Type	Description	Default
expression	str	Constraint equation, e.g. 'biso_Co2 = biso_Co1' or 'occ_Ba = 1 - occ_La'.	required

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

def create(self, *, expression: str) -> None:
    """
    Create a constraint from an expression string.

    Parameters
    ----------
    expression : str
        Constraint equation, e.g. ``'biso_Co2 = biso_Co1'`` or
        ``'occ_Ba = 1 - occ_La'``.
    """
    item = Constraint()
    item.expression = expression
    self.add(item)

factory

Constraints factory — delegates entirely to FactoryBase.

ConstraintsFactory

Bases: FactoryBase

Create constraint collections by tag.

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

class ConstraintsFactory(FactoryBase):
    """Create constraint collections by tag."""

    _default_rules = {
        frozenset(): 'default',
    }

fit_mode

enums

Enumeration for fit-mode values.

FitModeEnum

Bases: str, Enum

Fitting strategy for the analysis.

Source code in src/easydiffraction/analysis/categories/fit_mode/enums.py

class FitModeEnum(str, Enum):
    """Fitting strategy for the analysis."""

    SINGLE = 'single'
    JOINT = 'joint'

    @classmethod
    def default(cls) -> FitModeEnum:
        """Return the default fit mode (SINGLE)."""
        return cls.SINGLE

    def description(self) -> str:
        """Return a human-readable description of this fit mode."""
        if self is FitModeEnum.SINGLE:
            return 'Independent fitting of each experiment; no shared parameters'
        elif self is FitModeEnum.JOINT:
            return 'Simultaneous fitting of all experiments; some parameters are shared'

default() classmethod

Return the default fit mode (SINGLE).

Source code in src/easydiffraction/analysis/categories/fit_mode/enums.py

@classmethod
def default(cls) -> FitModeEnum:
    """Return the default fit mode (SINGLE)."""
    return cls.SINGLE

description()

Return a human-readable description of this fit mode.

Source code in src/easydiffraction/analysis/categories/fit_mode/enums.py

def description(self) -> str:
    """Return a human-readable description of this fit mode."""
    if self is FitModeEnum.SINGLE:
        return 'Independent fitting of each experiment; no shared parameters'
    elif self is FitModeEnum.JOINT:
        return 'Simultaneous fitting of all experiments; some parameters are shared'

factory

Fit-mode factory — delegates entirely to FactoryBase.

FitModeFactory

Bases: FactoryBase

Create fit-mode category items by tag.

Source code in src/easydiffraction/analysis/categories/fit_mode/factory.py

class FitModeFactory(FactoryBase):
    """Create fit-mode category items by tag."""

    _default_rules = {
        frozenset(): 'default',
    }

fit_mode

Fit-mode category item.

Stores the active fitting strategy as a CIF-serializable descriptor validated by FitModeEnum.

FitMode

Bases: CategoryItem

Fitting strategy selector.

Holds a single mode descriptor whose value is one of FitModeEnum members ('single' or 'joint').

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

@FitModeFactory.register
class FitMode(CategoryItem):
    """
    Fitting strategy selector.

    Holds a single ``mode`` descriptor whose value is one of
    ``FitModeEnum`` members (``'single'`` or ``'joint'``).
    """

    type_info = TypeInfo(
        tag='default',
        description='Fit-mode category',
    )

    def __init__(self) -> None:
        super().__init__()

        self._mode: StringDescriptor = StringDescriptor(
            name='mode',
            description='Fitting strategy',
            value_spec=AttributeSpec(
                default=FitModeEnum.default().value,
                validator=MembershipValidator(allowed=[member.value for member in FitModeEnum]),
            ),
            cif_handler=CifHandler(names=['_analysis.fit_mode']),
        )

        self._identity.category_code = 'fit_mode'

    @property
    def mode(self) -> StringDescriptor:
        """
        Fitting strategy.

        Reading this property returns the underlying
        ``StringDescriptor`` object. Assigning to it updates the
        parameter value.
        """
        return self._mode

    @mode.setter
    def mode(self, value: str) -> None:
        self._mode.value = value

mode property writable

Fitting strategy.

Reading this property returns the underlying StringDescriptor object. Assigning to it updates the parameter value.

joint_fit_experiments

default

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.

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

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

    def __init__(self) -> None:
        super().__init__()

        self._id: StringDescriptor = StringDescriptor(
            name='id',  # TODO: need new name instead of id
            description='Experiment identifier',  # TODO
            value_spec=AttributeSpec(
                default='_',
                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='Weight factor',  # TODO
            value_spec=AttributeSpec(
                default=0.0,
                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)

    # ------------------------------------------------------------------
    #  Public properties
    # ------------------------------------------------------------------

    @property
    def id(self) -> StringDescriptor:
        """
        Experiment identifier.

        Reading this property returns the underlying
        ``StringDescriptor`` object. Assigning to it updates the
        parameter value.
        """
        return self._id

    @id.setter
    def id(self, value: str) -> None:
        self._id.value = value

    @property
    def weight(self) -> NumericDescriptor:
        """
        Weight factor.

        Reading this property returns the underlying
        ``NumericDescriptor`` object. Assigning to it updates the
        parameter value.
        """
        return self._weight

    @weight.setter
    def weight(self, value: float) -> None:
        self._weight.value = value

id property writable

Experiment identifier.

Reading this property returns the underlying StringDescriptor object. Assigning to it updates the parameter value.

weight property writable

Weight factor.

Reading this property returns the underlying NumericDescriptor object. Assigning to it updates the parameter value.

JointFitExperiments

Bases: CategoryCollection

Collection of :class:JointFitExperiment items.

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

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

    type_info = TypeInfo(
        tag='default',
        description='Joint-fit experiment weights',
    )

    def __init__(self) -> None:
        """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/default.py

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

factory

Joint-fit-experiments factory — delegates to FactoryBase.

JointFitExperimentsFactory

Bases: FactoryBase

Create joint-fit experiment collections by tag.

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

class JointFitExperimentsFactory(FactoryBase):
    """Create joint-fit experiment collections by tag."""

    _default_rules = {
        frozenset(): 'default',
    }

fit_helpers

metrics

calculate_r_factor(y_obs, y_calc)

Calculate the 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	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 between observed and calculated data.

    Parameters
    ----------
    y_obs : np.ndarray
        Observed data points.
    y_calc : np.ndarray
        Calculated data points.

    Returns
    -------
    float
        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.

    Parameters
    ----------
    y_obs : np.ndarray
        Observed data points.
    y_calc : np.ndarray
        Calculated data points.

    Returns
    -------
    float
        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.

    Parameters
    ----------
    y_obs : np.ndarray
        Observed data points.
    y_calc : np.ndarray
        Calculated data points.

    Returns
    -------
    float
        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.

    Parameters
    ----------
    residuals : np.ndarray
        Residuals between observed and calculated data.
    num_parameters : int
        Number of free parameters used in the model.

    Returns
    -------
    float
        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 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 weighted R-factor between observed and calculated data.

    Parameters
    ----------
    y_obs : np.ndarray
        Observed data points.
    y_calc : np.ndarray
        Calculated data points.
    weights : np.ndarray
        Weights for each data point.

    Returns
    -------
    float
        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(structures, experiments)

Collect observed and calculated data for reliability calculations.

Parameters:

Name	Type	Description	Default
structures	Structures	Collection of structures.	required
experiments	Experiments	Collection of experiments.	required

Returns:

Type	Description
ndarray	Observed values.
ndarray	Calculated values.
Optional[ndarray]	Error values, or None if not available.

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

def get_reliability_inputs(
    structures: Structures,
    experiments: Experiments,
) -> Tuple[np.ndarray, np.ndarray, Optional[np.ndarray]]:
    """
    Collect observed and calculated data for reliability calculations.

    Parameters
    ----------
    structures : Structures
        Collection of structures.
    experiments : Experiments
        Collection of experiments.

    Returns
    -------
    np.ndarray
        Observed values.
    np.ndarray
        Calculated values.
    Optional[np.ndarray]
        Error values, or None if not available.
    """
    y_obs_all = []
    y_calc_all = []
    y_err_all = []
    for experiment in experiments.values():
        for structure in structures:
            structure._update_categories()
        experiment._update_categories()

        y_calc = experiment.data.intensity_calc
        y_meas = experiment.data.intensity_meas
        y_meas_su = experiment.data.intensity_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[object]] = None,
        chi_square: Optional[float] = None,
        reduced_chi_square: Optional[float] = None,
        message: str = '',
        iterations: int = 0,
        engine_result: Optional[object] = None,
        starting_parameters: Optional[List[object]] = None,
        fitting_time: Optional[float] = None,
        **kwargs: object,
    ) -> None:
        """
        Initialize FitResults with the given parameters.

        Parameters
        ----------
        success : bool, default=False
            Indicates if the fit was successful.
        parameters : Optional[List[object]], default=None
            List of parameters used in the fit.
        chi_square : Optional[float], default=None
            Chi-square value of the fit.
        reduced_chi_square : Optional[float], default=None
            Reduced chi-square value of the fit.
        message : str, default=''
            Message related to the fit.
        iterations : int, default=0
            Number of iterations performed.
        engine_result : Optional[object], default=None
            Result from the fitting engine.
        starting_parameters : Optional[List[object]], default=None
            Initial parameters for the fit.
        fitting_time : Optional[float], default=None
            Time taken for the fitting process.
        **kwargs : object
            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[object] = 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[object] = engine_result
        self.result: Optional[object] = None
        self.starting_parameters: List[object] = (
            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.

        Parameters
        ----------
        y_obs : Optional[List[float]], default=None
            Observed intensities for pattern R-factor metrics.
        y_calc : Optional[List[float]], default=None
            Calculated intensities for pattern R-factor metrics.
        y_err : Optional[List[float]], default=None
            Standard deviations of observed intensities for wR.
        f_obs : Optional[List[float]], default=None
            Observed structure-factor magnitudes for Bragg R.
        f_calc : Optional[List[float]], default=None
            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 or ''
            )  # 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[object]]	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[object]	Result from the fitting engine.	None
starting_parameters	Optional[List[object]]	Initial parameters for the fit.	None
fitting_time	Optional[float]	Time taken for the fitting process.	None
**kwargs	object	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[object]] = None,
    chi_square: Optional[float] = None,
    reduced_chi_square: Optional[float] = None,
    message: str = '',
    iterations: int = 0,
    engine_result: Optional[object] = None,
    starting_parameters: Optional[List[object]] = None,
    fitting_time: Optional[float] = None,
    **kwargs: object,
) -> None:
    """
    Initialize FitResults with the given parameters.

    Parameters
    ----------
    success : bool, default=False
        Indicates if the fit was successful.
    parameters : Optional[List[object]], default=None
        List of parameters used in the fit.
    chi_square : Optional[float], default=None
        Chi-square value of the fit.
    reduced_chi_square : Optional[float], default=None
        Reduced chi-square value of the fit.
    message : str, default=''
        Message related to the fit.
    iterations : int, default=0
        Number of iterations performed.
    engine_result : Optional[object], default=None
        Result from the fitting engine.
    starting_parameters : Optional[List[object]], default=None
        Initial parameters for the fit.
    fitting_time : Optional[float], default=None
        Time taken for the fitting process.
    **kwargs : object
        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[object] = 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[object] = engine_result
    self.result: Optional[object] = None
    self.starting_parameters: List[object] = (
        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.

    Parameters
    ----------
    y_obs : Optional[List[float]], default=None
        Observed intensities for pattern R-factor metrics.
    y_calc : Optional[List[float]], default=None
        Calculated intensities for pattern R-factor metrics.
    y_err : Optional[List[float]], default=None
        Standard deviations of observed intensities for wR.
    f_obs : Optional[List[float]], default=None
        Observed structure-factor magnitudes for Bragg R.
    f_calc : Optional[List[float]], default=None
        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 or ''
        )  # 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._verbosity: VerbosityEnum = VerbosityEnum.FULL

        self._df_rows: List[List[str]] = []
        self._display_handle: Optional[object] = None
        self._live: Optional[object] = 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.

        Parameters
        ----------
        residuals : np.ndarray
            Residuals between measured and calculated data.
        parameters : List[float]
            Current free parameters being fitted.

        Returns
        -------
        np.ndarray
            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.

        Parameters
        ----------
        minimizer_name : str
            Name of the minimizer used for the run.
        """
        if self._verbosity is VerbosityEnum.SILENT:
            return
        if self._verbosity is VerbosityEnum.SHORT:
            return

        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.

        Parameters
        ----------
        row : List[str]
            Columns corresponding to DEFAULT_HEADERS.
        """
        self._df_rows.append(row)
        if self._verbosity is not VerbosityEnum.FULL:
            return
        # Append and update via the active handle (Jupyter or
        # terminal live)
        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)

        if self._verbosity is not VerbosityEnum.FULL:
            return

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

    Parameters
    ----------
    row : List[str]
        Columns corresponding to DEFAULT_HEADERS.
    """
    self._df_rows.append(row)
    if self._verbosity is not VerbosityEnum.FULL:
        return
    # Append and update via the active handle (Jupyter or
    # terminal live)
    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)

    if self._verbosity is not VerbosityEnum.FULL:
        return

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

    Parameters
    ----------
    minimizer_name : str
        Name of the minimizer used for the run.
    """
    if self._verbosity is VerbosityEnum.SILENT:
        return
    if self._verbosity is VerbosityEnum.SHORT:
        return

    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.

    Parameters
    ----------
    residuals : np.ndarray
        Residuals between measured and calculated data.
    parameters : List[float]
        Current free parameters being fitted.

    Returns
    -------
    np.ndarray
        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') -> None:
        self.selection: str = selection
        self.engine: str = selection
        self.minimizer = MinimizerFactory.create(selection)
        self.results: Optional[FitResults] = None

    def fit(
        self,
        structures: Structures,
        experiments: Experiments,
        weights: Optional[np.array] = None,
        analysis: object = None,
        verbosity: VerbosityEnum = VerbosityEnum.FULL,
    ) -> None:
        """
        Run the fitting process.

        This method performs the optimization but does not display
        results. Use :meth:`show_fit_results` on the Analysis object to
        display the fit results after fitting is complete.

        Parameters
        ----------
        structures : Structures
            Collection of structures.
        experiments : Experiments
            Collection of experiments.
        weights : Optional[np.array], default=None
            Optional weights for joint fitting.
        analysis : object, default=None
            Optional Analysis object to update its categories during
            fitting.
        verbosity : VerbosityEnum, default=VerbosityEnum.FULL
            Console output verbosity.
        """
        params = structures.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:
            """
            Evaluate the residual for the current minimizer parameters.

            Parameters
            ----------
            engine_params : Dict[str, Any]
                Parameter values provided by the minimizer engine.

            Returns
            -------
            np.ndarray
                Residual array passed back to the minimizer.
            """
            return self._residual_function(
                engine_params=engine_params,
                parameters=params,
                structures=structures,
                experiments=experiments,
                weights=weights,
                analysis=analysis,
            )

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

    def _process_fit_results(
        self,
        structures: Structures,
        experiments: Experiments,
    ) -> None:
        """
        Collect reliability inputs and display fit results.

        This method is typically called by
        :meth:`Analysis.show_fit_results` rather than directly. It
        calculates R-factors and other metrics, then renders them to the
        console.

        Parameters
        ----------
        structures : Structures
            Collection of structures.
        experiments : Experiments
            Collection of experiments.
        """
        y_obs, y_calc, y_err = get_reliability_inputs(
            structures,
            experiments,
        )

        # 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,
        structures: Structures,
        experiments: Experiments,
    ) -> List[Parameter]:
        """
        Collect free parameters from structures and experiments.

        Parameters
        ----------
        structures : Structures
            Collection of structures.
        experiments : Experiments
            Collection of experiments.

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

    def _residual_function(
        self,
        engine_params: Dict[str, Any],
        parameters: List[Parameter],
        structures: Structures,
        experiments: Experiments,
        weights: Optional[np.array] = None,
        analysis: object = None,
    ) -> np.ndarray:
        """
        Compute residuals between measured and calculated patterns.

        It updates the parameter values according to the
        optimizer-provided engine_params.

        Parameters
        ----------
        engine_params : Dict[str, Any]
            Engine-specific parameter dict.
        parameters : List[Parameter]
            List of parameters being optimized.
        structures : Structures
            Collection of structures.
        experiments : Experiments
            Collection of experiments.
        weights : Optional[np.array], default=None
            Optional weights for joint fitting.
        analysis : object, default=None
            Optional Analysis object to update its categories during
            fitting.

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

        # Update categories to reflect new parameter values
        # Order matters: structures first (symmetry, structure),
        # then analysis (constraints), then experiments (calculations)
        for structure in structures:
            structure._update_categories()

        if analysis is not None:
            analysis._update_categories(called_by_minimizer=True)

        # 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):
            # Update experiment-specific calculations
            experiment._update_categories(called_by_minimizer=True)

            # Calculate the difference between measured and calculated
            # patterns
            y_calc: np.ndarray = experiment.data.intensity_calc
            y_meas: np.ndarray = experiment.data.intensity_meas
            y_meas_su: np.ndarray = experiment.data.intensity_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(structures, experiments, weights=None, analysis=None, verbosity=VerbosityEnum.FULL)

Run the fitting process.

This method performs the optimization but does not display results. Use :meth:show_fit_results on the Analysis object to display the fit results after fitting is complete.

Parameters:

Name	Type	Description	Default
structures	Structures	Collection of structures.	required
experiments	Experiments	Collection of experiments.	required
weights	Optional[array]	Optional weights for joint fitting.	None
analysis	object	Optional Analysis object to update its categories during fitting.	None
verbosity	VerbosityEnum	Console output verbosity.	VerbosityEnum.FULL

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

def fit(
    self,
    structures: Structures,
    experiments: Experiments,
    weights: Optional[np.array] = None,
    analysis: object = None,
    verbosity: VerbosityEnum = VerbosityEnum.FULL,
) -> None:
    """
    Run the fitting process.

    This method performs the optimization but does not display
    results. Use :meth:`show_fit_results` on the Analysis object to
    display the fit results after fitting is complete.

    Parameters
    ----------
    structures : Structures
        Collection of structures.
    experiments : Experiments
        Collection of experiments.
    weights : Optional[np.array], default=None
        Optional weights for joint fitting.
    analysis : object, default=None
        Optional Analysis object to update its categories during
        fitting.
    verbosity : VerbosityEnum, default=VerbosityEnum.FULL
        Console output verbosity.
    """
    params = structures.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:
        """
        Evaluate the residual for the current minimizer parameters.

        Parameters
        ----------
        engine_params : Dict[str, Any]
            Parameter values provided by the minimizer engine.

        Returns
        -------
        np.ndarray
            Residual array passed back to the minimizer.
        """
        return self._residual_function(
            engine_params=engine_params,
            parameters=params,
            structures=structures,
            experiments=experiments,
            weights=weights,
            analysis=analysis,
        )

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

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,
        verbosity: VerbosityEnum = VerbosityEnum.FULL,
    ) -> None:
        """
        Initialize progress tracking and timer.

        Parameters
        ----------
        minimizer_name : str
            Human-readable name shown in progress.
        verbosity : VerbosityEnum, default=VerbosityEnum.FULL
            Console output verbosity.
        """
        self.tracker.reset()
        self.tracker._verbosity = verbosity
        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.

        Parameters
        ----------
        parameters : List[Any]
            List of free parameters to be fitted.

        Returns
        -------
        Dict[str, Any]
            Mapping of keyword arguments to pass into ``_run_solver``.
        """
        pass

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

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

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

        Parameters
        ----------
        parameters : List[object]
            Parameters after the solver finished.
        raw_result : object
            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: object) -> bool:
        """Determine whether the fit was successful."""
        pass

    def fit(
        self,
        parameters: List[object],
        objective_function: Callable[..., object],
        verbosity: VerbosityEnum = VerbosityEnum.FULL,
    ) -> FitResults:
        """
        Run the full minimization workflow.

        Parameters
        ----------
        parameters : List[object]
            Free parameters to optimize.
        objective_function : Callable[..., object]
            Callable returning residuals for a given set of engine
            arguments.
        verbosity : VerbosityEnum, default=VerbosityEnum.FULL
            Console output verbosity.

        Returns
        -------
        FitResults
            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, verbosity=verbosity)

        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, object],
        parameters: List[object],
        structures: object,
        experiments: object,
        calculator: object,
    ) -> np.ndarray:
        """Default objective helper computing residuals array."""
        return self._compute_residuals(
            engine_params,
            parameters,
            structures,
            experiments,
            calculator,
        )

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

fit(parameters, objective_function, verbosity=VerbosityEnum.FULL)

Run the full minimization workflow.

Parameters:

Name	Type	Description	Default
parameters	List[object]	Free parameters to optimize.	required
objective_function	Callable[..., object]	Callable returning residuals for a given set of engine arguments.	required
verbosity	VerbosityEnum	Console output verbosity.	VerbosityEnum.FULL

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[object],
    objective_function: Callable[..., object],
    verbosity: VerbosityEnum = VerbosityEnum.FULL,
) -> FitResults:
    """
    Run the full minimization workflow.

    Parameters
    ----------
    parameters : List[object]
        Free parameters to optimize.
    objective_function : Callable[..., object]
        Callable returning residuals for a given set of engine
        arguments.
    verbosity : VerbosityEnum, default=VerbosityEnum.FULL
        Console output verbosity.

    Returns
    -------
    FitResults
        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, verbosity=verbosity)

    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 DFO-LS (derivative-free least-squares).

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

@MinimizerFactory.register
class DfolsMinimizer(MinimizerBase):
    """Minimizer using DFO-LS (derivative-free least-squares)."""

    type_info = TypeInfo(
        tag='dfols',
        description='DFO-LS derivative-free least-squares optimization',
    )

    def __init__(
        self,
        name: str = 'dfols',
        max_iterations: int = DEFAULT_MAX_ITERATIONS,
        **kwargs: object,
    ) -> 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[object]) -> Dict[str, object]:
        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: object, **kwargs: object) -> object:
        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[object],
        raw_result: object,
    ) -> None:
        """
        Synchronize the solver result back to the parameters.

        Parameters
        ----------
        parameters : List[object]
            List of parameters being optimized.
        raw_result : object
            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):
            # Bypass validation but set the dirty flag so
            # _update_categories() knows work is needed.
            param._set_value_from_minimizer(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: object) -> bool:
        """
        Determine success from DFO-LS result dictionary.

        Parameters
        ----------
        raw_result : object
            The result object returned by the solver.

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

factory

Minimizer factory — delegates to FactoryBase.

MinimizerFactory

Bases: FactoryBase

Factory for creating minimizer instances.

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

class MinimizerFactory(FactoryBase):
    """Factory for creating minimizer instances."""

    _default_rules = {
        frozenset(): 'lmfit',
    }

lmfit

LmfitMinimizer

Bases: MinimizerBase

Minimizer using the lmfit package.

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

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

    type_info = TypeInfo(
        tag='lmfit',
        description='LMFIT with Levenberg-Marquardt least squares',
    )

    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[object],
    ) -> Dict[str, object]:
        """
        Prepare the solver arguments for the lmfit minimizer.

        Parameters
        ----------
        parameters : List[object]
            List of parameters to be optimized.

        Returns
        -------
        Dict[str, object]
            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: object, **kwargs: object) -> object:
        """
        Run the lmfit solver.

        Parameters
        ----------
        objective_function : object
            The objective function to minimize.
        **kwargs : object
            Additional arguments for the solver.

        Returns
        -------
        object
            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[object],
        raw_result: object,
    ) -> None:
        """
        Synchronize the result from the solver to the parameters.

        Parameters
        ----------
        parameters : List[object]
            List of parameters being optimized.
        raw_result : object
            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:
                # Bypass validation but set the dirty flag so
                # _update_categories() knows work is needed.
                param._set_value_from_minimizer(param_result.value)
                param.uncertainty = getattr(param_result, 'stderr', None)

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

        Parameters
        ----------
        raw_result : object
            The result object returned by the solver.

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

    def _iteration_callback(
        self,
        params: lmfit.Parameters,
        iter: int,
        resid: object,
        *args: object,
        **kwargs: object,
    ) -> None:
        """
        Handle each iteration callback of the minimizer.

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