python-workflow-definition/example_workflows/nfdi/workflow.py at main · pythonworkflow/python-workflow-definition · GitHub

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
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
import os
from conda_subprocess import check_output
import shutil


def generate_mesh(domain_size: float, source_directory: str) -> str:
    stage_name = "preprocessing"
    gmsh_output_file_name = "square.msh"
    source_file_name ="unit_square.geo"
    os.makedirs(stage_name, exist_ok=True)
    _copy_file_from_source(stage_name=stage_name, source_file_name=source_file_name, source_directory=source_directory)
    _ = check_output(
        [
            "gmsh", "-2", "-setnumber", "domain_size", str(domain_size),
            source_file_name, "-o", gmsh_output_file_name
        ],
        prefix_name=stage_name,
        cwd=stage_name,
        universal_newlines=True,
    ).split("\n")
    return os.path.abspath(os.path.join(stage_name, gmsh_output_file_name))


def convert_to_xdmf(gmsh_output_file : str) -> dict:
    stage_name = "preprocessing"
    meshio_output_file_name = "square.xdmf"
    os.makedirs(stage_name, exist_ok=True)
    _copy_file(stage_name=stage_name, source_file=gmsh_output_file)
    _ = check_output(
        ["meshio", "convert", os.path.basename(gmsh_output_file), meshio_output_file_name],
        prefix_name=stage_name,
        cwd=stage_name,
        universal_newlines=True,
    ).split("\n")
    return {
        "xdmf_file": os.path.abspath(os.path.join(stage_name, meshio_output_file_name)),
        "h5_file": os.path.join(os.path.abspath(stage_name), "square.h5"),
    }


def poisson(meshio_output_xdmf: str, meshio_output_h5: str, source_directory: str) -> dict:
    stage_name = "processing"
    poisson_output_pvd_file_name = "poisson.pvd"
    poisson_output_numdofs_file_name = "numdofs.txt"
    source_file_name = "poisson.py"
    os.makedirs(stage_name, exist_ok=True)
    _copy_file_from_source(stage_name=stage_name, source_file_name=source_file_name, source_directory=source_directory)
    _copy_file(stage_name=stage_name, source_file=meshio_output_xdmf)
    _copy_file(stage_name=stage_name, source_file=meshio_output_h5)
    _ = check_output(
        [
            "python", "poisson.py", "--mesh", os.path.basename(meshio_output_xdmf), "--degree", "2",
            "--outputfile", poisson_output_pvd_file_name, "--num-dofs", poisson_output_numdofs_file_name
        ],
        prefix_name=stage_name,
        cwd=stage_name,
        universal_newlines=True,
    ).split("\n")
    return {
        "numdofs": _poisson_collect_output(numdofs_file=os.path.join(stage_name, poisson_output_numdofs_file_name)),
        "pvd_file": os.path.abspath(os.path.join(stage_name, poisson_output_pvd_file_name)),
        "vtu_file": os.path.abspath(os.path.join(stage_name, "poisson000000.vtu")),
    }


def plot_over_line(poisson_output_pvd_file: str, poisson_output_vtu_file: str, source_directory: str) -> str:
    stage_name = "postprocessing"
    pvbatch_output_file_name = "plotoverline.csv"
    source_file_name = "postprocessing.py"
    os.makedirs(stage_name, exist_ok=True)
    _copy_file_from_source(stage_name=stage_name, source_file_name=source_file_name, source_directory=source_directory)
    _copy_file(stage_name=stage_name, source_file=poisson_output_pvd_file)
    _copy_file(stage_name=stage_name, source_file=poisson_output_vtu_file)
    _ = check_output(
        ["pvbatch", source_file_name, os.path.basename(poisson_output_pvd_file), pvbatch_output_file_name],
        prefix_name=stage_name,
        cwd=stage_name,
        universal_newlines=True,
    ).split("\n")
    return os.path.abspath(os.path.join("postprocessing", pvbatch_output_file_name))


def substitute_macros(pvbatch_output_file: str, ndofs: int, domain_size: float, source_directory: str) -> str:
    stage_name = "postprocessing"
    source_file_name = "prepare_paper_macros.py"
    template_file_name = "macros.tex.template"
    macros_output_file_name = "macros.tex"
    os.makedirs(stage_name, exist_ok=True)
    _copy_file_from_source(stage_name=stage_name, source_file_name=source_file_name, source_directory=source_directory)
    _copy_file_from_source(stage_name=stage_name, source_file_name=template_file_name, source_directory=source_directory)
    _copy_file(stage_name=stage_name, source_file=pvbatch_output_file)
    _ = check_output(
        [
            "python", "prepare_paper_macros.py", "--macro-template-file", template_file_name,
            "--plot-data-path", os.path.basename(pvbatch_output_file), "--domain-size", str(domain_size),
            "--num-dofs", str(ndofs), "--output-macro-file", macros_output_file_name,
        ],
        prefix_name=stage_name,
        cwd=stage_name,
        universal_newlines=True,
    ).split("\n")
    return os.path.abspath(os.path.join(stage_name, macros_output_file_name))


def compile_paper(macros_tex: str, plot_file: str, source_directory: str) -> str:
    stage_name = "postprocessing"
    paper_output = "paper.pdf"
    source_file_name = "paper.tex"
    os.makedirs(stage_name, exist_ok=True)
    _copy_file_from_source(stage_name=stage_name, source_file_name=source_file_name, source_directory=source_directory)
    _copy_file(stage_name=stage_name, source_file=macros_tex)
    _copy_file(stage_name=stage_name, source_file=plot_file)
    _ = check_output(
        ["tectonic", source_file_name],
        prefix_name=stage_name,
        universal_newlines=True,
        cwd=stage_name,
    ).split("\n")
    return os.path.abspath(os.path.join(stage_name, paper_output))


def _poisson_collect_output(numdofs_file: str) -> int:
    with open(os.path.join(numdofs_file), "r") as f:
        return int(f.read())


def _copy_file(stage_name: str, source_file: str):
    input_file = os.path.join(os.path.abspath(stage_name), os.path.basename(source_file))
    if input_file != source_file:
        shutil.copyfile(source_file, input_file)


def _copy_file_from_source(stage_name: str, source_file_name: str, source_directory: str):
    source_file = os.path.join(source_directory, source_file_name)
    shutil.copyfile(source_file, os.path.join(stage_name, source_file_name))