From 47734100f9142cc3879273f103cf46265d783d22 Mon Sep 17 00:00:00 2001 From: ymzhang0 Date: Thu, 2 Jul 2026 17:08:30 +0200 Subject: [PATCH 1/2] feat(supercon): add automatic temps setting from Allen-Dynes Tc --- src/aiida_epw/tools/workchain.py | 31 +++++++++++++++++++++++++++++ src/aiida_epw/workflows/supercon.py | 3 +++ tests/tools/test_workchain.py | 30 ++++++++++++++++++++++++++++ 3 files changed, 64 insertions(+) diff --git a/src/aiida_epw/tools/workchain.py b/src/aiida_epw/tools/workchain.py index 5dfc9d8a..c061b215 100644 --- a/src/aiida_epw/tools/workchain.py +++ b/src/aiida_epw/tools/workchain.py @@ -342,3 +342,34 @@ def get_default_target_basepath(computer): else: raise ValueError(f"Unsupported transport type: {computer.transport_type}") return target_basepath + + +def set_auto_temps(inputs, last_conv_calc): + """Automatically set EPW calculation temps parameter from Allen-Dynes critical temperature of a previous calculation. + + If `temps` is not specified or is None in inputs.parameters['INPUTEPW'], + it sets `nstemp` to 10 and `temps` to be a range between 0.5 * Tc and 2.0 * Tc, + using the Allen_Dynes_Tc output of the `last_conv_calc`. + """ + from aiida.orm import Dict + + # Safe access to inputs.parameters + parameters = inputs.parameters.get_dict() if hasattr(inputs, "parameters") else {} + inputepw = parameters.setdefault("INPUTEPW", {}) + + if "temps" not in inputepw or inputepw.get("temps") is None: + try: + allen_dynes_tc = last_conv_calc.outputs.output_parameters["Allen_Dynes_Tc"] + except (AttributeError, KeyError) as exc: + raise ValueError( + "Could not find `Allen_Dynes_Tc` in the output parameters of the last convergence calculation." + ) from exc + + tmin = 0.5 * allen_dynes_tc + tmax = 2.0 * allen_dynes_tc + + inputepw["nstemp"] = 10 + inputepw["temps"] = f"{tmin:.4f} {tmax:.4f}" + + # Write back to inputs.parameters + inputs.parameters = Dict(parameters) diff --git a/src/aiida_epw/workflows/supercon.py b/src/aiida_epw/workflows/supercon.py index 6cd7edde..f1182ab1 100644 --- a/src/aiida_epw/workflows/supercon.py +++ b/src/aiida_epw/workflows/supercon.py @@ -11,6 +11,7 @@ from aiida_epw.calculations.epw import serialize_restart_type from aiida_epw.workflows.base import EpwBaseWorkChain from aiida_epw.data import A2fData +from aiida_epw.tools.workchain import set_auto_temps from aiida.engine import calcfunction @@ -462,6 +463,7 @@ def run_final_epw_iso(self): inputs = AttributeDict( self.exposed_inputs(EpwBaseWorkChain, namespace="epw_final_iso") ) + set_auto_temps(inputs, self.ctx.epw_interp[-1]) inputs.structure = self.inputs.structure parent_folder_epw = self.ctx.epw_interp[-1].outputs.remote_folder @@ -500,6 +502,7 @@ def run_final_epw_aniso(self): inputs = AttributeDict( self.exposed_inputs(EpwBaseWorkChain, namespace="epw_final_aniso") ) + set_auto_temps(inputs, self.ctx.epw_interp[-1]) inputs.structure = self.inputs.structure parent_folder_epw = self.ctx.epw_interp[-1].outputs.remote_folder diff --git a/tests/tools/test_workchain.py b/tests/tools/test_workchain.py index ad7710de..c265d603 100644 --- a/tests/tools/test_workchain.py +++ b/tests/tools/test_workchain.py @@ -218,3 +218,33 @@ def get_configuration(self): ssh_computer = MockSshComputer() assert get_default_target_basepath(ssh_computer) == "/remote/testuser/stash" + + +def test_set_auto_temps(): + """Test set_auto_temps correctly parses Allen_Dynes_Tc and configures input parameters.""" + from aiida_epw.tools.workchain import set_auto_temps + from aiida.orm import Dict + + # 1. Mock inputs and last_conv_calc + inputs = SimpleNamespace(parameters=Dict(dict={"INPUTEPW": {}})) + + last_conv_calc = SimpleNamespace( + outputs=SimpleNamespace(output_parameters={"Allen_Dynes_Tc": 12.0}) + ) + + # 2. Run set_auto_temps + set_auto_temps(inputs, last_conv_calc) + + # 3. Assert outputs + params = inputs.parameters.get_dict() + assert params["INPUTEPW"]["nstemp"] == 10 + assert params["INPUTEPW"]["temps"] == "6.0000 24.0000" + + # 4. Assert that if temps is already defined, it is not overwritten + inputs_predefined = SimpleNamespace( + parameters=Dict(dict={"INPUTEPW": {"temps": "5.0000 45.0000", "nstemp": 40}}) + ) + set_auto_temps(inputs_predefined, last_conv_calc) + params_predefined = inputs_predefined.parameters.get_dict() + assert params_predefined["INPUTEPW"]["nstemp"] == 40 + assert params_predefined["INPUTEPW"]["temps"] == "5.0000 45.0000" From 8f910e7ed02c2df82283120991b0934796da58fb Mon Sep 17 00:00:00 2001 From: ymzhang0 Date: Wed, 19 Aug 2026 14:38:07 +0200 Subject: [PATCH 2/2] refine set temperature function --- src/aiida_epw/tools/workchain.py | 27 ++++++++------------------- src/aiida_epw/workflows/supercon.py | 19 ++++++++++++++----- tests/tools/test_workchain.py | 28 ++++++++++------------------ 3 files changed, 32 insertions(+), 42 deletions(-) diff --git a/src/aiida_epw/tools/workchain.py b/src/aiida_epw/tools/workchain.py index c061b215..1e3d5471 100644 --- a/src/aiida_epw/tools/workchain.py +++ b/src/aiida_epw/tools/workchain.py @@ -344,32 +344,21 @@ def get_default_target_basepath(computer): return target_basepath -def set_auto_temps(inputs, last_conv_calc): +def set_auto_temps( + parameters, allen_dynes_tc, lower_bound=0.5, upper_bound=2.0, nstemp=10 +): """Automatically set EPW calculation temps parameter from Allen-Dynes critical temperature of a previous calculation. If `temps` is not specified or is None in inputs.parameters['INPUTEPW'], - it sets `nstemp` to 10 and `temps` to be a range between 0.5 * Tc and 2.0 * Tc, - using the Allen_Dynes_Tc output of the `last_conv_calc`. + it sets `nstemp` to 10 and `temps` to be a range between 0.5 * Tc and 2.0 * Tc. """ - from aiida.orm import Dict - - # Safe access to inputs.parameters - parameters = inputs.parameters.get_dict() if hasattr(inputs, "parameters") else {} inputepw = parameters.setdefault("INPUTEPW", {}) if "temps" not in inputepw or inputepw.get("temps") is None: - try: - allen_dynes_tc = last_conv_calc.outputs.output_parameters["Allen_Dynes_Tc"] - except (AttributeError, KeyError) as exc: - raise ValueError( - "Could not find `Allen_Dynes_Tc` in the output parameters of the last convergence calculation." - ) from exc - - tmin = 0.5 * allen_dynes_tc - tmax = 2.0 * allen_dynes_tc + tmin = lower_bound * allen_dynes_tc + tmax = upper_bound * allen_dynes_tc - inputepw["nstemp"] = 10 + inputepw["nstemp"] = nstemp inputepw["temps"] = f"{tmin:.4f} {tmax:.4f}" - # Write back to inputs.parameters - inputs.parameters = Dict(parameters) + return parameters diff --git a/src/aiida_epw/workflows/supercon.py b/src/aiida_epw/workflows/supercon.py index f1182ab1..d275ba32 100644 --- a/src/aiida_epw/workflows/supercon.py +++ b/src/aiida_epw/workflows/supercon.py @@ -463,7 +463,10 @@ def run_final_epw_iso(self): inputs = AttributeDict( self.exposed_inputs(EpwBaseWorkChain, namespace="epw_final_iso") ) - set_auto_temps(inputs, self.ctx.epw_interp[-1]) + allen_dynes_tc = self.ctx.epw_interp[-1].outputs.output_parameters[ + "Allen_Dynes_Tc" + ] + new_parameters = set_auto_temps(inputs.parameters.get_dict(), allen_dynes_tc) inputs.structure = self.inputs.structure parent_folder_epw = self.ctx.epw_interp[-1].outputs.remote_folder @@ -474,9 +477,9 @@ def run_final_epw_iso(self): inputs.restart_type = serialize_restart_type("ephread") if self.ctx.degaussq: - parameters = inputs.parameters.get_dict() - parameters["INPUTEPW"]["degaussq"] = self.ctx.degaussq - inputs.parameters = orm.Dict(parameters) + new_parameters["INPUTEPW"]["degaussq"] = self.ctx.degaussq + + inputs.parameters = orm.Dict(new_parameters) inputs.metadata.call_link_label = "epw_final_iso" @@ -502,7 +505,11 @@ def run_final_epw_aniso(self): inputs = AttributeDict( self.exposed_inputs(EpwBaseWorkChain, namespace="epw_final_aniso") ) - set_auto_temps(inputs, self.ctx.epw_interp[-1]) + + allen_dynes_tc = self.ctx.epw_interp[-1].outputs.output_parameters[ + "Allen_Dynes_Tc" + ] + new_parameters = set_auto_temps(inputs.parameters.get_dict(), allen_dynes_tc) inputs.structure = self.inputs.structure parent_folder_epw = self.ctx.epw_interp[-1].outputs.remote_folder @@ -512,6 +519,8 @@ def run_final_epw_aniso(self): inputs.restart_type = serialize_restart_type("ephread") + inputs.parameters = orm.Dict(new_parameters) + inputs.metadata.call_link_label = "epw_final_aniso" workchain_node = self.submit(EpwBaseWorkChain, **inputs) self.report( diff --git a/tests/tools/test_workchain.py b/tests/tools/test_workchain.py index c265d603..06259081 100644 --- a/tests/tools/test_workchain.py +++ b/tests/tools/test_workchain.py @@ -223,28 +223,20 @@ def get_configuration(self): def test_set_auto_temps(): """Test set_auto_temps correctly parses Allen_Dynes_Tc and configures input parameters.""" from aiida_epw.tools.workchain import set_auto_temps - from aiida.orm import Dict - # 1. Mock inputs and last_conv_calc - inputs = SimpleNamespace(parameters=Dict(dict={"INPUTEPW": {}})) - - last_conv_calc = SimpleNamespace( - outputs=SimpleNamespace(output_parameters={"Allen_Dynes_Tc": 12.0}) - ) + # 1. Mock parameters and tc + parameters = {"INPUTEPW": {}} + allen_dynes_tc = 12.0 # 2. Run set_auto_temps - set_auto_temps(inputs, last_conv_calc) + result = set_auto_temps(parameters, allen_dynes_tc) # 3. Assert outputs - params = inputs.parameters.get_dict() - assert params["INPUTEPW"]["nstemp"] == 10 - assert params["INPUTEPW"]["temps"] == "6.0000 24.0000" + assert result["INPUTEPW"]["nstemp"] == 10 + assert result["INPUTEPW"]["temps"] == "6.0000 24.0000" # 4. Assert that if temps is already defined, it is not overwritten - inputs_predefined = SimpleNamespace( - parameters=Dict(dict={"INPUTEPW": {"temps": "5.0000 45.0000", "nstemp": 40}}) - ) - set_auto_temps(inputs_predefined, last_conv_calc) - params_predefined = inputs_predefined.parameters.get_dict() - assert params_predefined["INPUTEPW"]["nstemp"] == 40 - assert params_predefined["INPUTEPW"]["temps"] == "5.0000 45.0000" + params_predefined = {"INPUTEPW": {"temps": "5.0000 45.0000", "nstemp": 40}} + result_predefined = set_auto_temps(params_predefined, allen_dynes_tc) + assert result_predefined["INPUTEPW"]["nstemp"] == 40 + assert result_predefined["INPUTEPW"]["temps"] == "5.0000 45.0000"