Commit 2e3fc510 authored by Andrei Sobolev's avatar Andrei Sobolev
Browse files

Merge branch 'fhi-vibes' into 'master'

Add Spectroscopy workflow to Control Generator Step

See merge request !64
parents b91e2ae5 12a0655d
Loading
Loading
Loading
Loading
Loading
+1 −1
Original line number Diff line number Diff line
@@ -15,7 +15,7 @@ cache:
    - node_modules/

variables:
  PYTHON_VERSION: "3.8"
  PYTHON_VERSION: "3.10"
  DARWIN_AMD64_BINARY: "gims-${CI_COMMIT_TAG}-darwin.app"
  LINUX_AMD64_BINARY: "gims-${CI_COMMIT_TAG}-linux.x"
  WINDOWS_AMD64_BINARY: "gims-${CI_COMMIT_TAG}-windows.exe"
+24 −29
Original line number Diff line number Diff line
@@ -9,48 +9,47 @@ from gims.structure import Structure


class Exciting(CodeInputs):
    def __init__(self, data, input_dir, species_dir, session=None):
    def __init__(self, data, species_dir, session=None):
        """Exciting Code class

        Provides the specific methods to prepare the corresponding input files.

        Parameters:

        data: directory
        data: dict
            Dictionary generated from form sent by client.
        input_dir: string
            Path to the directory, where input can be written to.
        species_dir: string
            Path the exciting species.
        species_dir: str|Path
            Path to the exciting species defaults directory.
        """
        super().__init__(data, species_dir, session)

        super(Exciting, self).__init__(session)
        self.info = {"references": [], "DownloadInputFilesPage": {}}
        if self.structure is None:
            is_periodic = 'nkgrid' in self.flat_form
            self.structure = Structure.from_form(self.flat_form, is_periodic)

        if "structure" in data:
            structure = Structure.from_dict(data["structure"])
        else:
            is_periodic = "nkgrid" in data["form"]
            structure = Structure.from_form(data["form"], is_periodic)

        self.calc_from_form(structure, data["form"], species_dir)
        os.makedirs(os.path.dirname(input_dir))
        structure.calc.dir = input_dir
        structure.calc.write(structure)
    def write_inputs(self, input_dir):
        """ Write input files to a given directory
        Args:
            input_dir (pathlib.Path): input directory
        """
        self.calc_from_form(self.structure, self.flat_form, self.species_dir)
        os.makedirs(input_dir, exist_ok=True)
        self.structure.calc.dir = input_dir
        self.structure.calc.write(self.structure)

        if not "structure" in data:
        if not self._has_structure:
            self.insert_comment(
                "This is just a dummy structure! Please modify it!",
                "structure",
                os.path.join(input_dir, "input.xml"),
            )

        species = self.get_species(structure)
        species = self.get_species(self.structure)
        for s in species:
            fname = s + ".xml"
            shutil.copyfile(
                os.path.join(species_dir, fname),
                os.path.join(structure.calc.dir, fname),
                os.path.join(self.species_dir, fname),
                os.path.join(self.structure.calc.dir, fname),
            )

        self.error = None
@@ -67,7 +66,6 @@ class Exciting(CodeInputs):
        for s in structure.get_chemical_symbols():
            if s not in species:
                species.append(s)

        return species

    @staticmethod
@@ -78,9 +76,7 @@ class Exciting(CodeInputs):
        doc = ET.parse(filename)
        crystal = doc.getroot().find(element)
        crystal.insert(0, ET.Comment(comment))

        doc.write(filename)
        # print(os.listdir(os.path.dirname(filename)))

    def calc_from_form(self, struct, form_dict, species_dir):
        """Generates the ASE calculator from the control-generator form.
@@ -89,8 +85,8 @@ class Exciting(CodeInputs):

        struct: ASE atoms object

        formDict: JSON object
            Dictionary of the control generator form.
        form_dict: dict
            Flat dictionary of control-generator flags.

        species_dir: String (Path)
            Path to the Species root dir.
@@ -136,7 +132,7 @@ class Exciting(CodeInputs):
        calc = Exciting(speciespath=".", paramdict=paramdict)

        if "bandStructure" in form_dict:
            self.prepare_band_input(struct.cell, calc, density=int(form_dict[key]))
            self.prepare_band_input(struct.cell, calc, density=int(form_dict["bandStructure"]))

        struct.set_calculator(calc)

@@ -153,7 +149,6 @@ class Exciting(CodeInputs):
        density: int
            Number of kpoints per Angstrom. Default: 20
        """

        self.get_band_path_info(cell)
        bp = cell.bandpath()
        r_kpts = resolve_kpt_path_string(bp.path, bp.special_points)
+160 −0
Original line number Diff line number Diff line
import ast
import configparser
from pathlib import Path

from ase.io import write as ase_write

from gims.inputs.codeinputs import CodeInputs


README = """\
This calculation should be run with FHI-vibes. 

First, make sure that FHI-Vibes is installed on your resources. If not, follow the following link 
for installation instructions:
               https://vibes-developers.gitlab.io/vibes/Installation/

Second, run FHI-Vibes with the following command:
```
vibes run phonopy vibes.in
```
or 
```
vibes run md vibes.in 
```
depending on your choice of phonon picture: harmonic (phonopy) or unharmonic (MD).

After the calculation is finished, you can upload the output directory to GIMS to analyze outputs.
"""

class FHIVibes(CodeInputs):
    """Writes vibes.in (INI format) + geometry.in for FHI-vibes phonopy/MD workflows.

    Frontend form sections are mapped to vibes.in INI sections as follows:
      - "phonopy``  → [phonopy]
      - "md_vibes`` → [md]
      - All other sections → [calculator.parameters]
      - "basisSettings`` key → [calculator.basissets] default value
    """

    # Frontend section labels that become their own INI sections (name → vibes.in section)
    _VIBES_SECTION_MAP = {
        'phonopy': "_write_phonopy",
        'md_vibes': "_write_md",
        'polarization': '_write_polarization',
        'dfpt': '_write_raman'
    }

    def write_inputs(self, input_dir):
        input_dir = Path(input_dir)
        self._write_geometry(input_dir)
        self._write_vibes_in(input_dir)
        with (input_dir / 'README.md').open('w') as f:
            f.write(README)

    def _write_geometry(self, input_dir: Path):
        ase_write(str(input_dir / 'geometry.in'), self.structure, format='aims')

    def _get_vibes_in(self):
        config = configparser.ConfigParser()
        config.optionxform = str  # preserve key case — FHI-aims flags are case-sensitive

        config['files'] = {'geometry': 'geometry.in'}
        config['calculator'] = {'name': 'aims', 'socketio': 'False'}

        # default parameters
        calc_params = {'relativistic': 'atomic_zora scalar'}
        basis_set = ''

        for section_label, values in self.form.items():
            if section_label == 'spectroscopy':
                continue  # no real keys, only managing section
            if section_label in self._VIBES_SECTION_MAP:
                continue  # handled separately below
            for key, value in values.items():
                if key == 'basisSettings':
                    basis_set = value
                elif key == 'species':
                    continue
                else:
                    calc_params[key] = self._fmt_aims(value)

        if not basis_set:
            raise ValueError('No basis set specified')

        if calc_params:
            config['calculator.parameters'] = calc_params
        config['calculator.basissets'] = {'default': basis_set}

        for section_label, method_name in self._VIBES_SECTION_MAP.items():
            if section_label in self.form:
                config.read_dict(getattr(self, method_name)(self.form[section_label]))
        return config

    def _write_vibes_in(self, input_dir: Path):
        config = self._get_vibes_in()
        lines = []
        for section, values in config.items():
            if section == "DEFAULT":
                continue  # default section for ConfigParser
            lines.append(f'[{section}]')
            for key, value in values.items():
                if 'output' in key:  # one line per entry: "output: <type> <i> <v1> <v2> ..."
                    output_type = key.removeprefix('output.')
                    entries = ast.literal_eval(value)
                    for i, entry in enumerate(entries, 1):
                        vals = ' '.join(str(x) for x in entry) if isinstance(entry, (list, tuple)) else entry
                        lines.append(f'output: {output_type} {i} {vals}')
                else:
                    lines.append(f'{key}: {value}')
            lines.append('')
        (input_dir / 'vibes.in').write_text('\n'.join(lines))

    @staticmethod
    def _fmt_aims(value) -> str:
        """Format a value for [calculator.parameters]: lists become space-separated integers."""
        if isinstance(value, (list, tuple)):
            return ' '.join(str(v) for v in value)
        return str(value)

    @staticmethod
    def _fmt_vibes(value) -> str:
        """Format a value for vibes-native sections: lists keep Python list notation."""
        if isinstance(value, (list, tuple)):
            return str(list(value))
        return str(value)

    def _write_phonopy(self, config):
        section = {
            "phonopy": {
                "supercell_matrix": self._fmt_vibes([1, 1, 1]),
                "is_diagonal": False,
                "q_mesh": self._fmt_vibes([1, 1, 1]),
                "workdir": "phonopy"
            }}

        for k, v in config.items():
            section['phonopy'][k] = self._fmt_vibes(v)
        return section

    def _write_md(self, config):
        md_params = config.pop('MD_params')
        section = {'md': {'workdir': 'md'},
                   'md.kwargs': {'logfile': 'md.log',
                                 'temperature': self._fmt_vibes(md_params[0]),
                                 'friction': self._fmt_vibes(md_params[1])}}
        for k, v in config.items():
            section['md'][k] = self._fmt_vibes(v)
        return section

    def _write_raman(self, config):
        section = {'calculator.parameters': {'DFPT': 'dielectric'}}
        for k, v in config.items():
            section['calculator.parameters'][k] = self._fmt_vibes(v)
        return section

    @staticmethod
    def _write_polarization(config):
        v = config.pop('output_polarization')
        assert not config, f"Unexpected keys in polarization section: {config.keys()}"
        return {'calculator.parameters': {'output.polarization': [v[:3], v[3:6], v[6:]]}}
+45 −54
Original line number Diff line number Diff line
@@ -4,7 +4,7 @@ from pathlib import Path
import numpy as np

from ase import __version__ as ase_version
from ase.calculators.aims import Aims
from ase.calculators.aims import Aims, AimsProfile
from ase.calculators.calculator import kpts2mp
from ase.dft.kpoints import resolve_kpt_path_string, kpoint_convert
from gims.inputs.codeinputs import CodeInputs
@@ -24,57 +24,46 @@ class FHIaims(CodeInputs):

        data (dict):
            Dictionary generated from form sent by client.
            ``data['form']`` is the two-level ``{section_label: {flag: value}}`` dict
            produced by the frontend.
        species_dir (str|Path):
            Path to the FHI-aims species defaults directory.
        """
        super(FHIaims, self).__init__(session)
        self.info = {"references": [], "DownloadInputFilesPage": {}}
        self._has_structure = "structure" in data
        self.species_dir = species_dir
        self.calc = None
        self.error = None
        super().__init__(data, species_dir, session)

        if self._has_structure:
            self.structure = Structure.from_dict(data["structure"])
            if np.any(self.structure.get_initial_magnetic_moments()):
                data["form"]["spin"] = "collinear"
        else:
            # make dummy structure object
            is_periodic = any([x in data["form"] for x in ("k_grid", "k_grid_density")])
            self.structure = Structure.from_form(data["form"], is_periodic)
        # Work from a flat copy of the two-level form
        control = self.flat_form.copy()

        # Band structure calculations with HSE06+SOC should have exx_band_structure_version set; #89
        if ('hse06' in data['form']['xc']
                and 'include_spin_orbit' in data['form']
                and 'bandStructure' in data['form']):
            data['form']['exx_band_structure_version'] = '1'
        if self.structure is None:
            is_periodic = any(x in control for x in ('k_grid', 'k_grid_density'))
            self.structure = Structure.from_form(control, is_periodic)
        elif np.any(self.structure.get_initial_magnetic_moments()):
            control['spin'] = 'collinear'

        if "relativistic" not in data["form"]:
            data["form"]["relativistic"] = "atomic_zora scalar"
        if 'relativistic' not in control:
            control['relativistic'] = 'atomic_zora scalar'

        self.is_gw_workflow = 'needDFTInputs' in data['form']
        self.is_gw_workflow = self.workflow == 'GW'
        if self.is_gw_workflow:
            data['form'].pop('needDFTInputs')
            control.pop('needDFTInputs', None)

        # MD inputs
        self.is_md = 'MD_run' in data['form']
        if self.is_md:
            time = data['form'].pop('MD_run_time')
            ensemble = data['form']['MD_run']
            params = data['form'].pop('MD_run_params', [])
            data['form']['MD_run'] = [time, ensemble] + params
        if self.workflow == 'MD':
            time = control.pop('MD_run_time')
            ensemble = control['MD_run']
            params = control.pop('MD_run_params', [])
            control['MD_run'] = [time, ensemble] + params

        self.control = data['form']
        self.control = control

    def set_structure(self, structure):
        """ Changes the structure for the inputs without rebuilding all the inputs
        Args:
            structure (Structure): a structure to set
        """
        self.structure = structure
        self.control["species"] = list(set(self.structure.get_chemical_symbols()))
        super().set_structure(structure)
        self.control['species'] = list(set(self.structure.get_chemical_symbols()))
        if np.any(self.structure.get_initial_magnetic_moments()):
            self.control["spin"] = "collinear"
            self.control['spin'] = 'collinear'

    def set_control(self, key, value):
        """ Sets control generator key to a predefined value
@@ -89,18 +78,25 @@ class FHIaims(CodeInputs):
        Args:
            input_dir (pathlib.Path): input directory
        """
        # get the function object
        # vdW Tkatchenko-Scheffler is unsupported for alkali species — swap the key
        if 'vdw_correction_hirshfeld' in self.control:
            alkali = {'Li', 'Na', 'K', 'Rb', 'Cs', 'Fr'}
            if set(self.structure.get_chemical_symbols()).intersection(alkali):
                del self.control['vdw_correction_hirshfeld']
                self.control['vdw_correction_hirshfeld_alkali'] = True

        calc_keys = self.calc_from_form(self.structure, self.control, self.species_dir.as_posix())
        if self.is_gw_workflow:
            self._write_inputs(input_dir / 'GW', calc_keys)
            calc_keys_dft = {k: v for (k, v) in calc_keys.items() if k not in self.gw_keys}
            self._write_inputs(input_dir / 'DFT', calc_keys_dft)
            with open(input_dir / "README.txt", "w") as f:
                s = """\
    After all calculations are finished, you can archive everything in the directory with the following `tar` command:
        $ tar cvzf workflow.tar.gz */
    and provide the resultant gzip archive to the Output Analyzer"""
                f.write(s)
                f.write(
                    "After all calculations are finished, you can archive everything in the "
                    "directory with the following `tar` command:\n"
                    "    $ tar cvzf workflow.tar.gz */\n"
                    "and provide the resultant gzip archive to the Output Analyzer"
                )
        else:
            self._write_inputs(input_dir, calc_keys)

@@ -116,27 +112,26 @@ class FHIaims(CodeInputs):
            input_dir (pathlib.Path): input directory
            calc_keys (dict): a dict of cal control keys
        """
        # setup Aims calculation
        self.calc = Aims()
        self.calc = Aims(profile=AimsProfile(
            command='aims.x',
            default_species_directory=str(self.species_dir),
        ))
        self.calc.directory = Path(input_dir)
        try:
            self.calc.template.write_input(self.calc.directory, self.structure, calc_keys, [])
            self.calc.template.write_input(self.calc.profile, self.calc.directory, self.structure, calc_keys, [])
            (self.calc.directory / "parameters.ase").unlink(missing_ok=True)
            if not self._has_structure:
                # delete dummy object
                (self.calc.directory / "geometry.in").unlink(missing_ok=True)
        except (FileNotFoundError, RuntimeError):
            self.error = "FileNotFoundError"

    def _write_inputs_stable(self, input_dir, calc_keys):
        """ A subroutine used with current version of ASE (3.22)
        """ A subroutine used with the current version of ASE (3.22)
        Args:
            input_dir (pathlib.Path): input directory
            calc_keys (dict): a dict of cal control keys
        """
        # setup Aims calculation
        if 'k_grid_density' in calc_keys:
            # old ASE version, just calculate k_grid
            k_grid_density = calc_keys.pop('k_grid_density')
            calc_keys['k_grid'] = tuple(kpts2mp(self.structure, k_grid_density))

@@ -160,8 +155,8 @@ class FHIaims(CodeInputs):

        struct: ASE atoms object

        formDict: JSON object
            Dictionary of the control generator form.
        form_dict: dict
            Flat dictionary of control-generator flags.

        species_dir: String (Path)
            Path to the Species root dir.
@@ -171,12 +166,10 @@ class FHIaims(CodeInputs):
            if key == "basisSettings":
                calc_keys["species_dir"] = os.path.join(species_dir, form_dict[key])
            elif key in ("bandStructure", "GWBandStructure"):
                # print(key, formDict[key])
                calc_keys["output"] = self.prepare_band_input(struct.cell, density=int(form_dict[key]))
            elif key == "species":
                continue
            else:
                # Only use this if 'key' is already proper FHI-aims keyword
                calc_keys[key] = form_dict[key]
        return calc_keys

@@ -193,7 +186,6 @@ class FHIaims(CodeInputs):
        """
        self.get_band_path_info(cell)
        bp = cell.bandpath()
        # print(cell.get_bravais_lattice())
        r_kpts = resolve_kpt_path_string(bp.path, bp.special_points)

        lines_and_labels = []
@@ -201,7 +193,6 @@ class FHIaims(CodeInputs):
            dists = coords[1:] - coords[:-1]
            lengths = [np.linalg.norm(d) for d in kpoint_convert(cell, skpts_kc=dists)]
            points = np.int_(np.round(np.asarray(lengths) * density))
            # I store it here for now. Might be needed to get global info.
            lines_and_labels.append(
                [points, labels[:-1], labels[1:], coords[:-1], coords[1:]]
            )
+8 −0
Original line number Diff line number Diff line
import sys

from gims.inputs.FHIaims import FHIaims
from gims.inputs.FHIVibes import FHIVibes
from gims.inputs.Exciting import Exciting
from gims.inputs.codeinputs import CodeInputs

@@ -12,17 +13,24 @@ def inputs_cls(step_def):
        step_def (str|dict): a ControlGenerator step definition
    """
    code_name = None
    workflow = None
    if isinstance(step_def, str):
        code_name = step_def
    elif isinstance(step_def, dict):
        if 'control' in step_def:
            code_name = step_def['control']['code']
            workflow = step_def['control'].get('workflow')
        elif 'code' in step_def:
            code_name = step_def['code']
            workflow = step_def.get('workflow')

    if code_name is None:
        raise KeyError('Could not find the code name in the workflow step definition')

    # Route FHI-aims Spectroscopy workflow to FHI-vibes writer
    if code_name == 'FHIaims' and workflow == 'Spectroscopy':
        return FHIVibes

    try:
        inputs_cls_obj = getattr(sys.modules[__name__], code_name)
    except AttributeError as ex:
Loading