✨ Added CIFReflectionsMetadata

xtl.diffraction.reflections.metadata:CIFReflectionsMetadata
- Metadata container for reflections extracted from a CIF file
- The .from_gemmi() method allows instantiation directly from a
  gemmi.ReflnBlock object
- The .spec_lines attribute holds the GemmiCIF2MTZSpecs that were used
  during the internal conversion to a gemmi.Mtz object
- The .column_types_inferred attribute holds the MTZ column types that
  can be inferred from the mmCIF tags using the spec_lines, if present

xtl.diffraction.reflections.files:CIFReflectionsFile
- Refactored the .read() method to handle metadata addition to the
  resulting ReflectionsCollection
- Metadata are extracted from an intermediate gemmi.ReflnBlock
- The .read() method now also accepts an optional `spec_lines` argument
  for influencing the internal conversion to gemmi.Mtz - the default
  spec is used when not provided

pyproject.toml & requirements.txt
- Upgraded gemmi to 0.7.0
- Upgraded reciprocalspaceship to 1.0.4

Author: dtriand

pyproject.toml

@@ -35,7 +35,7 @@ dependencies = [
     "distro>=1.9.0",
     "fabio>=2023.6.0",
     "f90nml>=1.4.4",
-    "gemmi>=0.5.5",
+    "gemmi>=0.7.0",
     "hdf5plugin>=4.2.0",
     "matplotlib>=3.8.0",
     "numpy>=1.26.0",
@@ -44,7 +44,7 @@ dependencies = [
     "pydantic>=2.11.5",
     "pyfai>=2023.9.0,<2024.10",
     "pyxray>=1.7.0",
-    "reciprocalspaceship>=1.0.3",
+    "reciprocalspaceship>=1.0.4",
     "requests>=2.31.0",
     "rich>=13.6.0",
     "tabulate>=0.8.10",

requirements.txt

@@ -4,7 +4,7 @@ defusedxml==0.7.1
 distro==1.9.0
 fabio==2023.6.0
 f90nml==1.4.4
-gemmi==0.5.8
+gemmi==0.7.1
 hdf5plugin==4.2.0
 matplotlib==3.8.0
 numpy==1.26.0
@@ -13,7 +13,7 @@ pint==0.19.2
 pydantic==2.11.5
 pyfai==2023.9.0
 pyxray==1.7.0
-reciprocalspaceship==1.0.3
+reciprocalspaceship==1.0.4
 requests==2.31.0
 rich==13.9.4
 sphinx==8.1.3

src/xtl/common/options.py

@@ -49,7 +49,7 @@ def Option(
         deprecated: Deprecated | str | bool | None = _Unset,  # deprecation message
         # Validation
         validate_default: bool | None = _Unset,  # validate default value
-        choices: str | tuple[Any, ...] | list[Any] | set[Any] | None = _Unset,  # iterable of choices
+        choices: Iterable[Any] | None = _Unset,  # iterable of choices
         #  for numbers
         gt: SupportsGt | None = _Unset,  # greater than
         ge: SupportsGe | None = _Unset,  # greater than or equal to

src/xtl/common/validators.py

@@ -32,7 +32,7 @@ def validate_length(value: Any, length: int) -> Any:
     return value
 
 
-def validate_choices(value: Any, choices: str | tuple[Any]) -> Any:
+def validate_choices(value: Any, choices: Iterable[Any]) -> Any:
     """
     Check if the value is contained in the provided choices.
     """
@@ -116,7 +116,7 @@ def LengthValidator(length: int) -> AfterValidator:
     return AfterValidator(partial(validate_length, length=length))
 
 
-def ChoicesValidator(choices: str | tuple[Any, ...]) -> AfterValidator:
+def ChoicesValidator(choices: Iterable[Any]) -> AfterValidator:
     """
     `Pydantic` validator to check if a value is in the provided choices.
     """

src/xtl/diffraction/reflections/__init__.py

@@ -1,5 +1,4 @@
-__all__ = ['ReflectionsData', 'ReflectionsCollection', 'mtz_types']
+__all__ = ['ReflectionsData', 'ReflectionsCollection']
 
-from .mtz_types import mtz_types
 from .reflections import ReflectionsData, ReflectionsCollection
 

src/xtl/diffraction/reflections/files.py

@@ -2,7 +2,7 @@
 import warnings
 from enum import Enum
 from pathlib import Path
-from typing import Any, TYPE_CHECKING
+from typing import Any, Sequence, Type, TYPE_CHECKING
 
 import gemmi
 import reciprocalspaceship as rs
@@ -241,26 +241,33 @@ def sniff(file: str | Path) -> bool:
             # No reflection data found :(
             return False
 
-    def read(self, block_id: int | str = 0) -> 'ReflectionsCollection':
+    def read(self, block_id: int | str = 0,
+             spec_lines: Sequence[str | Type['GemmiCIF2MTZSpec']] = None) -> \
+            'ReflectionsCollection':
         """
         Read the CIF file and return a ReflectionsCollection. If multiple data blocks
         are present in the CIF file, the `block_id` parameter can be used to select
-        a specific one.
+        a specific one. Internally, the CIF data are converted to MTZ format using
+        gemmi. The conversion can be customized by providing specification lines.
 
         :param block_id: Index or name of the data block to read (default: 0).
+        :param spec_lines: Sequence of GemmiCIF2MTZSpec or strings to specify how to
+            convert CIF data to MTZ. If not provided, the default specifications
+            will be used.
         :raise IndexError: If the specified `block_id` does not exist in the CIF file.
         :raise TypeError: If `block_id` is not an int or str.
         """
         import gemmi
         import reciprocalspaceship as rs
         from xtl.diffraction.reflections import ReflectionsCollection
+        from xtl.diffraction.reflections.metadata import CIFReflectionsMetadata
 
         cif = gemmi.cif.read(str(self.file))
-        blocks = gemmi.as_refln_blocks(cif)
+        rblocks: gemmi.ReflnBlocks = gemmi.as_refln_blocks(cif)
 
         if isinstance(block_id, str):
             # Get the index of the block by name
-            block_names = [block.block.name for block in blocks]
+            block_names = [rblock.block.name for rblock in rblocks]
             if block_id not in block_names:
                 raise IndexError(f'Block {block_id!r} not found in CIF file. '
                                  f'Available blocks: {block_names}')
@@ -269,21 +276,55 @@ def read(self, block_id: int | str = 0) -> 'ReflectionsCollection':
             raise TypeError(f'\'block_id\' must be an int or str, got {type(block_id)}')
 
         # Check if the block_id is within the range of available blocks
-        if block_id >= len(blocks):
-            raise IndexError(f'File {self.file} has only {len(blocks)} blocks, '
+        if block_id < 0 or block_id >= len(rblocks):
+            raise IndexError(f'File {self.file} has only {len(rblocks)} blocks, '
                              f'but {block_id=} was requested.')
 
-        block = blocks[block_id]
-        mtz = gemmi.CifToMtz().convert_block_to_mtz(block)
+        rblock = rblocks[block_id]
+        # Check if the block contains reflections
+        if not rblock.default_loop:
+            raise ValueError(f'Block {block_id} in CIF file {self.file} '
+                             f'does not contain any reflections.')
+
+        if not spec_lines:
+            # Get default spec_lines
+            if rblock.is_merged():
+                spec_lines = GemmiSpecs.CIF2MTZ.merged
+            else:
+                spec_lines = GemmiSpecs.CIF2MTZ.unmerged
+        else:
+            # Cast provided spec_lines
+            lines = []
+            for spec in spec_lines:
+                if isinstance(spec, str):
+                    lines.append(GemmiCIF2MTZSpec.from_line(spec))
+                elif isinstance(spec, GemmiCIF2MTZSpec):
+                    lines.append(spec)
+                else:
+                    raise TypeError(f'Invalid spec type: {type(spec)}. Expected str or '
+                                    f'{GemmiCIF2MTZSpec.__class__.__name__}.')
+            spec_lines = tuple(lines)
+
+        # Convert the CIF block to MTZ
+        converter = gemmi.CifToMtz()
+        converter.spec_lines = [spec.line for spec in spec_lines]
+        mtz = converter.convert_block_to_mtz(rblock)
+
+        # Convert gemmi.Mtz to rs.DataSet
         ds = rs.io.from_gemmi(mtz)
 
-        return ReflectionsCollection.from_rs(dataset=ds)
+        # Create metadata
+        metadata = CIFReflectionsMetadata.from_gemmi(rblock, spec_lines=spec_lines)
+
+        return ReflectionsCollection.from_rs(dataset=ds, metadata=metadata)
+
 
 _mtz_summary = rs.summarize_mtz_dtypes(print_summary=False)
 MTZ_DTYPES = {_mtz_summary['MTZ Code'][i]: getattr(rs.dtypes, _mtz_summary['Class'][i])
               for i in range(_mtz_summary.shape[0])}
 """Dictionary of MTZ column types to their corresponding ``rs.MTZDtype class``"""
 
+
 MTZ_COLUMN_TYPES = set(MTZ_DTYPES)
 """Set of valid column types according to the MTZ specification."""
 

src/xtl/diffraction/reflections/metadata.py

@@ -1,10 +1,20 @@
-from typing import Any
+__all__ = [
+    'ReflectionsMetadata',
+    'MTZBatchMetadata',
+    'MTZDatasetMetadata',
+    'MTZReflectionsMetadata',
+    'CIFReflectionsMetadata',
+    'ReflectionsMetadataType'
+]
+
+from typing import Any, Sequence, overload
 
 import gemmi
+from pydantic import computed_field
 
 from xtl.common.options import Option, Options
 from xtl.common.serializers import GemmiUnitCell, GemmiSpaceGroup, GemmiMat33
-from xtl.diffraction.reflections.files import ReflectionsFileType
+from xtl.diffraction.reflections.files import ReflectionsFileType, GemmiCIF2MTZSpec
 
 
 class ReflectionsMetadata(Options):
@@ -151,7 +161,7 @@ class MTZReflectionsMetadata(ReflectionsMetadata):
                                                           'missing values (VALM)')
     history: tuple[str, ...] | None = Option(default=None, desc='History lines')
 
-    @property
+    @computed_field(description='Whether the reflections are merged or unmerged')
     def is_merged(self) -> bool | None:
         """
         Check if the reflections are merged or unmerged. The presence of batches in the
@@ -185,23 +195,79 @@ def from_gemmi(cls, mtz: gemmi.Mtz):
         )
 
 
-class CIFDatasetMetadata(Options):
-    wavelength: float | None = Option(default=None, gt=0.0,
-                                      desc='Wavelength in Angstroms')
-    unit_cell: gemmi.UnitCell | None = Option(default=None, formatter=GemmiUnitCell,
-                                              desc='Unit cell parameters in '
-                                                   'Angstroms/degrees')
+class CIFReflectionsMetadata(ReflectionsMetadata):
+    """
+    Metadata for reflections extracted from a CIF file.
+    """
 
+    origin_file_type: ReflectionsFileType = Option(default=ReflectionsFileType.CIF,
+                                                   desc='Type of the file where the '
+                                                        'reflections originated from')
+    entry_id: str | None = Option(default=None, desc='Entry ID of the CIF block')
+    is_merged: bool | None = Option(default=None, desc='Whether the reflections are '
+                                                       'merged or unmerged')
+    spec_lines: tuple[GemmiCIF2MTZSpec, ...] | None = \
+        Option(default_factory=tuple,
+               formatter=lambda x: tuple(l.line for l in x),
+               desc='List of specification lines used during CIF to MTZ conversion')
+
+    @computed_field(description='Column types inferred from the column labels using '
+                                'the specification lines, if available')
+    def column_types_inferred(self) -> tuple[str | None, ...] | None:
+        """
+        Infer the column types from the column labels using the specification lines,
+        if available.
+        """
+        column_types = []
+        if not self.column_labels or not self.spec_lines:
+            return None
+
+        specs = list(self.spec_lines) + \
+                [GemmiCIF2MTZSpec.from_line(l) for l in
+                 ['index_h H H 0', 'index_k K H 0', 'index_l L H 0']]
+        for label in self.column_labels:
+            inferred = False
+            for spec in specs:
+                if label == spec.tag:
+                    column_types.append(spec.column_type)
+                    inferred = True
+                    break
+            if not inferred:
+                column_types.append(None)
+
+        return tuple(column_types)
+
+    @classmethod
+    def from_gemmi(cls, rblock: gemmi.ReflnBlock,
+                   spec_lines: Sequence[str | GemmiCIF2MTZSpec] = None):
+        """
+        Create an instance from a ``gemmi.ReflnBlock`` object.
+        """
+        if not isinstance(rblock, gemmi.ReflnBlock):
+            raise TypeError(f'Expected gemmi.ReflnBlock, got {type(rblock).__name__}')
+
+        resolution_high = rblock.block.find_value('_reflns.d_resolution_high')
+        resolution_low = rblock.block.find_value('_reflns.d_resolution_low')
+        n_obs = rblock.block.find_value('_reflns.number_obs')
+        resolution_high = float(resolution_high) if resolution_high else None
+        resolution_low = float(resolution_low) if resolution_low else None
+        n_obs = int(n_obs) if n_obs else None
+
+        return cls(
+            origin_file_type=ReflectionsFileType.CIF,
+            name=rblock.block.name,
+            entry_id=rblock.entry_id,
+            unit_cell=rblock.cell,
+            space_group=rblock.spacegroup,
+            wavelength=rblock.wavelength,
+            resolution_high=resolution_high,
+            resolution_low=resolution_low,
+            no_reflections=n_obs,
+            is_merged=rblock.is_merged(),
+            column_labels=tuple(rblock.column_labels()),
+            spec_lines=spec_lines or tuple()
+        )
 
-class CIFReflectionsMetadata(ReflectionsMetadata):
-    origin_file_type: ReflectionsFileType = Option(default=ReflectionsFileType.CIF)
-    datasets: list[CIFDatasetMetadata] | None = Option(default=None)
-    title: str | None = Option(default=None, max_length=64,
-                               desc='Title of the CIF file')
-    spec_lines: list[str] | None = Option(default=None,
-                                          desc='List of specification lines')
-    history: list[str] | None = Option(default=None,
-                                       desc='History of the CIF file')
 
 
 ReflectionsMetadataType = ReflectionsMetadata | MTZReflectionsMetadata | \