File size: 1,890 Bytes
9e04b81
 
6fdbe91
9241861
7471e3d
a133fcb
9241861
a133fcb
7471e3d
56200e6
 
9241861
 
4091f40
cd36c2d
7471e3d
 
4091f40
cd36c2d
4091f40
7471e3d
56200e6
 
 
7471e3d
 
 
 
 
 
 
 
 
cd36c2d
7471e3d
 
 
 
 
 
 
cd36c2d
6fdbe91
cd36c2d
9241861
 
 
 
a133fcb
7471e3d
a133fcb
cd36c2d
 
a133fcb
cd36c2d
 
 
 
 
9241861
cd36c2d
 
 
 
 
 
56200e6
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
import sys

import pytest
from ase.build import bulk
from prefect import flow
from prefect.testing.utilities import prefect_test_harness

from mlip_arena.models import MLIPEnum
from mlip_arena.tasks.eos import run as EOS
from mlip_arena.tasks.utils import get_calculator



@flow(persist_result=True)
def single_eos_flow(calculator_name, concurrent=True, cache=False):
    atoms = bulk("Cu", "fcc", a=3.6)

    return EOS.with_options(
        refresh_cache=not cache,
    )(
        atoms=atoms,
        calculator=get_calculator(
            calculator_name=calculator_name,
        ),
        optimizer="BFGSLineSearch",
        optimizer_kwargs=None,
        filter="FrechetCell",
        filter_kwargs=None,
        criterion=dict(
            fmax=0.1,
        ),
        max_abs_strain=0.1,
        npoints=6,
        concurrent=concurrent,
    )


@pytest.mark.skipif(
    sys.version_info[:2] != (3, 11),
    reason="avoid prefect race condition on concurrent tasks",
)
@pytest.mark.parametrize("concurrent", [False])
@pytest.mark.parametrize("model", [MLIPEnum["MACE-MP(M)"]])
def test_eos(model: MLIPEnum, concurrent: bool):
    """
    Test EOS prefect workflow with a simple cubic lattice.
    """

    with prefect_test_harness():
        result = single_eos_flow(
            calculator_name=model.name,
            concurrent=concurrent,
            cache=False,
        )
        assert isinstance(b0_scratch := result["b0"], float)

        # @pytest.mark.dependency(depends=["test_eos"])
        # @pytest.mark.parametrize("model", [MLIPEnum["MACE-MP(M)"]])
        # def test_eos_cache(model: MLIPEnum):

        result = single_eos_flow(
            calculator_name=model.name,
            concurrent=concurrent,
            cache=True,
        )
        assert isinstance(b0_cache := result["b0"], float)
        assert b0_scratch == pytest.approx(b0_cache, rel=1e-5)