Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 8 additions & 5 deletions src/aiida_epw/workflows/supercon.py
Original file line number Diff line number Diff line change
Expand Up @@ -174,6 +174,7 @@ def define(cls, spec):
cls.run_conv,
cls.inspect_conv,
),
cls.inspect_interpolation,
if_(cls.should_run_final)(
cls.run_final_epw_iso,
cls.inspect_final_epw_iso,
Expand Down Expand Up @@ -443,20 +444,22 @@ def inspect_conv(self):
frequency = workchain.outputs.a2f.get_array("frequency")
self.ctx.degaussq = frequency[-1] / 100

def should_run_final(self):
"""Check if the final EpwBaseWorkChain should be run."""
def inspect_interpolation(self):
"""Verify that the interpolation stage produced a usable result."""
if not self.ctx.epw_interp:
self.report(
"Allen-Dynes interpolation was not successful, epw_interp list is empty."
)
return self.exit_codes.ERROR_SUB_PROCESS_EPW_INTERP

if self.ctx.is_converged or self.inputs.always_run_final.value:
return True
else:
if not (self.ctx.is_converged or self.inputs.always_run_final.value):
self.report("Allen-Dynes Tc is not converged.")
return self.exit_codes.ERROR_ALLEN_DYNES_NOT_CONVERGED

def should_run_final(self):
"""Check if the final EpwBaseWorkChain should be run."""
return self.ctx.is_converged or self.inputs.always_run_final.value

def run_final_epw_iso(self):
"""Run the final EpwBaseWorkChain in isotropic mode."""
inputs = AttributeDict(
Expand Down
49 changes: 37 additions & 12 deletions tests/workflows/test_supercon.py
Original file line number Diff line number Diff line change
Expand Up @@ -151,18 +151,16 @@ def test_supercon_get_builder_from_protocol_base_failure(
)


def test_supercon_should_run_final():
"""Test should_run_final with different states of epw_interp and convergence."""
def test_supercon_inspect_interpolation():
"""Test interpolation inspection before final workchains are considered."""
from aiida_epw.workflows.supercon import SuperConWorkChain
from types import SimpleNamespace
from aiida.common.extendeddicts import AttributeDict

# Mock exit codes
class FakeExitCodes:
ERROR_SUB_PROCESS_EPW_INTERP = "ERROR_SUB_PROCESS_EPW_INTERP"
ERROR_ALLEN_DYNES_NOT_CONVERGED = "ERROR_ALLEN_DYNES_NOT_CONVERGED"

# Define fake workchain helper
def make_fake_workchain(epw_interp_list, is_converged, always_run_final):
reports = []
ctx = SimpleNamespace(
Expand All @@ -182,30 +180,57 @@ def make_fake_workchain(epw_interp_list, is_converged, always_run_final):
reports=reports,
)

# 1. epw_interp is empty -> should return ERROR_SUB_PROCESS_EPW_INTERP
wc1 = make_fake_workchain(
epw_interp_list=[], is_converged=True, always_run_final=True
)
assert SuperConWorkChain.should_run_final(wc1) == "ERROR_SUB_PROCESS_EPW_INTERP"
assert (
SuperConWorkChain.inspect_interpolation(wc1) == "ERROR_SUB_PROCESS_EPW_INTERP"
)
assert "empty" in wc1.reports[0]

# 2. epw_interp is not empty, converged -> should return True
wc2 = make_fake_workchain(
epw_interp_list=[object()], is_converged=True, always_run_final=False
)
assert SuperConWorkChain.should_run_final(wc2) is True
assert SuperConWorkChain.inspect_interpolation(wc2) is None

# 3. epw_interp is not empty, not converged but always_run_final is True -> should return True
wc3 = make_fake_workchain(
epw_interp_list=[object()], is_converged=False, always_run_final=True
)
assert SuperConWorkChain.should_run_final(wc3) is True
assert SuperConWorkChain.inspect_interpolation(wc3) is None

# 4. epw_interp is not empty, not converged, always_run_final is False -> should return ERROR_ALLEN_DYNES_NOT_CONVERGED
wc4 = make_fake_workchain(
epw_interp_list=[object()], is_converged=False, always_run_final=False
)
assert SuperConWorkChain.should_run_final(wc4) == "ERROR_ALLEN_DYNES_NOT_CONVERGED"
assert (
SuperConWorkChain.inspect_interpolation(wc4)
== "ERROR_ALLEN_DYNES_NOT_CONVERGED"
)
assert "not converged" in wc4.reports[0]


def test_supercon_should_run_final():
"""Test final-workchain gating after interpolation inspection."""
from aiida_epw.workflows.supercon import SuperConWorkChain
from types import SimpleNamespace
from aiida.common.extendeddicts import AttributeDict

def make_fake_workchain(is_converged, always_run_final):
ctx = SimpleNamespace(is_converged=is_converged)
inputs = AttributeDict(
{
"always_run_final": SimpleNamespace(value=always_run_final),
}
)
return SimpleNamespace(inputs=inputs, ctx=ctx)

wc1 = make_fake_workchain(is_converged=True, always_run_final=False)
assert SuperConWorkChain.should_run_final(wc1) is True

wc2 = make_fake_workchain(is_converged=False, always_run_final=True)
assert SuperConWorkChain.should_run_final(wc2) is True

wc3 = make_fake_workchain(is_converged=False, always_run_final=False)
assert SuperConWorkChain.should_run_final(wc3) is False


def test_epw_base_restart_types(fixture_code, generate_structure):
Expand Down
Loading