Note
Go to the end to download the full example code.
Create a discipline from an external executable#
from __future__ import annotations
import os
import subprocess
from typing import TYPE_CHECKING
from numpy import array
from gemseo import configure_logger
from gemseo.core.discipline import Discipline
if TYPE_CHECKING:
from gemseo.typing import StrKeyMapping
configure_logger()
<RootLogger root (INFO)>
Introduction#
Let's consider a binary software computing the float output
\(c = a^2 + b^2\) from two float inputs : 'a'
and 'b'
.
The inputs are read in the 'inputs.txt'
file
which looks like: a=1 b=2 and
the output is written to: 'outputs.txt'
which looks like c=5.
Then, the executable can be run using the shell command 'python run.py'
.
Let's make a discipline out of this from an initial 'inputs.txt'
.
Implementation of the discipline#
The construction of Discipline
consists in three steps:
Instantiate the
Discipline
using the super constructor,Initialize the grammars using the
JSONGrammar.update()
method,Set the default inputs from the initial
'inputs.txt'
The Discipline._run
method consists in three steps:
Get the input data from
Discipline.local_data
and write the'inputs.txt'
file,Run the executable using the
subprocess.run()
command (see more),Get the output values and store them to
Discipline.local_data
.
Now you can implement the discipline in the following way:
def parse_file(file_path):
data = {}
with open(file_path) as inf:
for line in inf:
if len(line) == 0:
continue
name, value = line.replace("\n", "").split("=")
data[name] = array([float(value)])
return data
def write_file(data, file_path) -> None:
with open(file_path, "w") as outf:
for name, value in list(data.items()):
outf.write(name + "=" + str(value[0]) + "\n")
class ShellExecutableDiscipline(Discipline):
def __init__(self) -> None:
super().__init__("ShellDisc")
self.input_grammar.update_from_names(["a", "b"])
self.output_grammar.update_from_names(["c"])
self.default_input_data = {"a": array([1.0]), "b": array([2.0])}
def _run(self, input_data: StrKeyMapping) -> StrKeyMapping | None:
cwd = os.getcwd()
inputs_file = os.path.join(cwd, "inputs.txt")
outputs_file = os.path.join(cwd, "outputs.txt")
write_file(input_data, inputs_file)
subprocess.run("python run.py".split(), cwd=cwd)
return parse_file(outputs_file)
Execution of the discipline#
Now we can run it with default input values:
shell_disc = ShellExecutableDiscipline()
shell_disc.execute()
{'a': array([1.]), 'b': array([2.]), 'c': array([5.])}
or run it with new input values:
shell_disc.execute({"a": array([2.0]), "b": array([3.0])})
{'a': array([2.]), 'b': array([3.]), 'c': array([13.])}
Total running time of the script: (0 minutes 0.127 seconds)