workflow for running magnetization

Author: unkcpz

src/aiida_sssp_workflow/protocol/magnetization.yml

@@ -0,0 +1,48 @@
+---
+standard:
+    name: acwf
+    description: Delta measure parameters compatible with AiiDA common workflow
+
+    occupations: smearing
+    degauss: 0.0045
+    smearing: fd
+    conv_thr_per_atom: 1.0e-9
+    kpoints_distance: 0.06
+    mixing_beta: 0.4
+
+balanced:
+    name: balanced
+    description: The balanced protocol from Gabriel
+
+    occupations: smearing
+    degauss: 0.02
+    smearing: fd
+    conv_thr_per_atom: 1.0e-8
+    kpoints_distance: 0.2
+    mixing_beta: 0.4
+
+test:
+    name: test-only
+    description: light loose set for test
+
+    occupations: smearing
+    degauss: 0.0045
+    smearing: fd
+    conv_thr_per_atom: 1.0e-6
+    kpoints_distance: 0.5
+    mixing_beta: 0.4
+    scale_count: 5
+    scale_increment: 0.02
+
+opsp:
+    name: opsp
+    description: For OPSP verifications
+
+    occupations: smearing
+    degauss: 0.0045
+    smearing: fd
+    conv_thr_per_atom: 1.0e-8
+    kpoints_distance: 0.06
+    mixing_beta: 0.4
+    scale_count: 5
+    scale_increment: 0.03

src/aiida_sssp_workflow/utils/structure.py

@@ -17,6 +17,15 @@
     ACWF_CONFIGURATIONS  # FIXME: should also have GS for bands and LANN
 )
 
+@calcfunction
+def scale_structure(
+    structure: orm.StructureData, scale_factor: orm.Float
+) -> orm.StructureData:
+    """Scale the structure with the given scaling factor."""
+    ase = structure.get_ase().copy()
+    ase.set_cell(ase.get_cell() * float(scale_factor) ** (1 / 3), scale_atoms=True)
+    return orm.StructureData(ase=ase)
+
 
 @calcfunction
 def get_default_configuration(element: orm.Str, property: orm.Str) -> orm.Str:

src/aiida_sssp_workflow/workflows/evaluate/_eos.py

@@ -4,23 +4,14 @@
 from typing import List
 
 from aiida import orm
-from aiida.engine import WorkChain, append_, calcfunction, run_get_node
+from aiida.engine import WorkChain, append_, run_get_node
 from aiida.plugins import CalculationFactory, WorkflowFactory
 
+from aiida_sssp_workflow.utils.structure import scale_structure
+
 PwBaseWorkChain = WorkflowFactory("quantumespresso.pw.base")
 birch_murnaghan_fit = CalculationFactory("sssp_workflow.birch_murnaghan_fit")
 
-
-@calcfunction
-def scale_structure(
-    structure: orm.StructureData, scale_factor: orm.Float
-) -> orm.StructureData:
-    """Scale the structure with the given scaling factor."""
-    ase = structure.get_ase().copy()
-    ase.set_cell(ase.get_cell() * float(scale_factor) ** (1 / 3), scale_atoms=True)
-    return orm.StructureData(ase=ase)
-
-
 class _EquationOfStateWorkChain(WorkChain):
     """Workflow to compute the equation of state for a given crystal structure."""
 

src/aiida_sssp_workflow/workflows/evaluate/_magnetization.py

@@ -0,0 +1,96 @@
+# -*- coding: utf-8 -*-
+"""
+A calcfunctian create_isolate_atom
+Create the structure of isolate atom
+"""
+
+from aiida import orm
+from aiida.engine import ToContext, calcfunction
+from aiida.plugins import DataFactory, WorkflowFactory
+
+from . import _BaseEvaluateWorkChain
+
+PwBaseWorkChain = WorkflowFactory("quantumespresso.pw.base")
+UpfData = DataFactory("pseudo.upf")
+
+
+@calcfunction
+def helper_get_hydrostatic_stress(output_trajectory, output_parameters):
+    """
+    doc
+    """
+    import numpy as np
+
+    output_stress = output_trajectory.get_array("stress")[0]
+    stress_unit = output_parameters["stress_units"]
+    hydrostatic_stress = np.trace(output_stress) / 3.0
+    return orm.Dict(
+        dict={
+            "stress_unit": stress_unit,
+            "hydrostatic_stress": hydrostatic_stress,
+        }
+    )
+
+
+class MagnetizationWorkChain(_BaseEvaluateWorkChain):
+    """WorkChain to calculate magnetization of input structure"""
+
+    @classmethod
+    def define(cls, spec):
+        """Define the process specification."""
+        # yapf: disable
+        super().define(spec)
+        spec.input('structure', valid_type=orm.StructureData,
+                    help='Ground state structure which the verification perform')
+        spec.input_namespace('pseudos', valid_type=UpfData, dynamic=True,
+                    help='A mapping of `UpfData` nodes onto the kind name to which they should apply.')
+        spec.expose_inputs(PwBaseWorkChain, exclude=["pw.structure", "pw.pseudos"])
+
+        spec.outline(
+            cls.run_scf,
+            cls.inspect_scf,
+            cls.finalize,
+        )
+        spec.output('output_parameters', valid_type=orm.Dict, required=True,
+                    help='The output parameters include pressure of the structure.')
+
+        # yapf: enable
+
+    def run_scf(self):
+        """
+        set the inputs and submit scf
+        """
+        inputs = self.exposed_inputs(PwBaseWorkChain)
+        inputs["pw"]["structure"] = self.inputs.structure
+        inputs["pw"]["pseudos"] = self.inputs.pseudos
+
+        running = self.submit(PwBaseWorkChain, **inputs)
+        self.report(f"Running pw scf calculation pk={running.pk}")
+
+        return ToContext(workchain_scf=running)
+
+    def inspect_scf(self):
+        """inspect the result of scf calculation."""
+        workchain = self.ctx.workchain_scf
+
+        if workchain.is_finished:
+            self._disable_cache(workchain)
+
+        if not workchain.is_finished_ok:
+            self.report(
+                f"PwBaseWorkChain for pressure evaluation failed with exit status {workchain.exit_status}"
+            )
+            return self.exit_codes.ERROR_SUB_PROCESS_FAILED_SCF
+
+        output_trajectory = workchain.outputs.output_trajectory
+        output_parameters = workchain.outputs.output_parameters
+
+        # Return the output parameters of current workchain
+        output_parameters = helper_get_hydrostatic_stress(
+            output_trajectory, output_parameters
+        )
+
+        self.out("output_parameters", output_parameters)
+
+    def finalize(self):
+        """set ecutwfc and ecutrho"""