from pathlib import Path
import sys
from typing import Optional, List, Any
from veloxchem.outputstream import OutputStream
from veloxchem.veloxchemlib import mpi_master
import mpi4py.MPI as MPI
from veloxchem.errorhandler import assert_msg_critical
import datetime
[docs]
class CifWriter:
"""Writer for crystallographic CIF files (Crystallographic Information File, .cif).
Handles CIF file creation in parallel MPI environments, with proper cell
and atom formatting for MOF and other crystalline data.
Attributes:
comm (Any): MPI communicator.
rank (int): MPI process rank.
nodes (int): Number of MPI processes.
ostream (OutputStream): Output stream for status messages or debugging.
filepath (Optional[Path]): Default path to write CIF files.
_debug (bool): Enables debug-level status messages if True.
cell_info (Optional[Any]): Placeholder for unit cell parameters.
supercell_boundary (Optional[Any]): Placeholder for supercell dimensions.
unit_cell (Optional[Any]): Placeholder for unit cell dimensions.
file_dir (Optional[Path]): Directory where the CIF will be written (set in write()).
Methods:
write: Writes a CIF file given cell, boundary, and atom data.
"""
def __init__(
self,
comm: Optional[Any] = None,
ostream: Optional[OutputStream] = None,
filepath: Optional[str] = None,
debug: bool = False
) -> None:
"""Initializes the CifWriter object.
Args:
comm (Any, optional): MPI communicator. Defaults to MPI.COMM_WORLD.
ostream (Optional[OutputStream]): Output stream for logging/debug info.
filepath (Optional[str]): Default path for CIF output.
debug (bool): Enable debug printing if True.
"""
if comm is None:
comm = MPI.COMM_WORLD
if ostream is None:
if comm.Get_rank() == mpi_master():
ostream = OutputStream(sys.stdout)
else:
ostream = OutputStream(None)
# mpi information
self.comm = comm
self.rank = self.comm.Get_rank()
self.nodes = self.comm.Get_size()
# output stream
self.ostream = ostream
self.filepath = filepath
self._debug = debug
# Placeholders for optional attributes
self.cell_info = None
self.supercell_boundary = None
self.unit_cell = None
self.file_dir = None
[docs]
def write(
self,
filepath: Optional[str] = None,
header: str = '',
lines: List[List[Any]] = [],
supercell_boundary: Optional[List[float]] = None,
cell_info: Optional[List[float]] = None,
) -> None:
"""Write data to a CIF file.
This method writes atomic, cell, and symmetry data to a new or existing
CIF file in the standard format. Used in the context of crystal structures
generated by mofbuilder.
Args:
filepath (Optional[str]): Output file path (.cif will be appended if missing).
header (str): Optional string describing the creation method.
lines (List[List[Any]]): Each sublist contains atom data in the form
[atom_type, atom_label, atom_number, residue_name, residue_number, x, y, z, spin, charge, note].
supercell_boundary (Optional[List[float]]): Max [x, y, z] of the supercell; used to normalize coordinates.
cell_info (Optional[List[float]]): Cell parameters [a, b, c, alpha, beta, gamma].
Raises:
AssertionError: If no output filepath is specified.
ValueError: If required parameters (lines, boundary, cell_info) are missing or malformed.
Example:
>>> writer = CifWriter()
>>> writer.write(
... filepath="my_structure.cif",
... header="MOFbuilder v1.0",
... lines=[["C", "C1", 1, "RES", 1, 1.0, 2.0, 3.0, 1.0, 0.0, ""]],
... supercell_boundary=[10.0,10.0,10.0],
... cell_info=[10.0,10.0,10.0,90.0,90.0,90.0]
... )
"""
filepath = Path(filepath) if filepath is not None else Path(self.filepath)
assert_msg_critical(filepath is not None, "ciffilepath is not specified")
self.file_dir = Path(filepath).parent
if self._debug:
self.ostream.print_info(f"targeting directory: {self.file_dir}")
self.file_dir.mkdir(parents=True, exist_ok=True)
if cell_info is None or supercell_boundary is None:
raise ValueError("Both cell_info and supercell_boundary must be provided")
a, b, c, alpha, beta, gamma = cell_info
x_max, y_max, z_max = map(float, supercell_boundary)
if filepath.suffix != ".cif":
filepath = filepath.with_suffix(".cif")
with open(filepath, 'w') as new_cif:
new_cif.write('data_' + filepath.name[:-4] + '\n')
new_cif.write('_audit_creation_date ' +
datetime.datetime.today().strftime('%Y-%m-%d') +
'\n')
new_cif.write("_audit_creation_method " + header + '\n')
new_cif.write("_symmetry_space_group_name 'P1'" + '\n')
new_cif.write('_symmetry_Int_Tables_number 1' + '\n')
new_cif.write('loop_' + '\n')
new_cif.write('_symmetry_equiv_pos_as_xyz' + '\n')
new_cif.write(' x,y,z' + '\n')
new_cif.write('_cell_length_a ' + str(a) + '\n')
new_cif.write('_cell_length_b ' + str(b) + '\n')
new_cif.write('_cell_length_c ' + str(c) + '\n')
new_cif.write('_cell_angle_alpha ' + str(alpha) +
'\n')
new_cif.write('_cell_angle_beta ' + str(beta) +
'\n')
new_cif.write('_cell_angle_gamma ' + str(gamma) +
'\n')
new_cif.write('loop_' + '\n')
new_cif.write('_atom_site_label' + '\n')
new_cif.write('_atom_site_type_symbol' + '\n')
new_cif.write('_atom_site_fract_x' + '\n')
new_cif.write('_atom_site_fract_y' + '\n')
new_cif.write('_atom_site_fract_z' + '\n')
for i, values in enumerate(lines):
atom_type = values[0]
atom_label = values[1]
x = float(values[5]) / x_max if x_max != 0 else 0
y = float(values[6]) / y_max if y_max != 0 else 0
z = float(values[7]) / z_max if z_max != 0 else 0
formatted_line = "%7s%4s%15.10f%15.10f%15.10f" % (
atom_type, atom_label, x, y, z)
new_cif.write(formatted_line + '\n')
new_cif.write('loop_' + '\n')