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:

  1. Instantiate the Discipline using the super constructor,

  2. Initialize the grammars using the JSONGrammar.update() method,

  3. Set the default inputs from the initial 'inputs.txt'

The Discipline._run method consists in three steps:

  1. Get the input data from Discipline.local_data and write the 'inputs.txt' file,

  2. Run the executable using the subprocess.run() command (see more),

  3. 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)

Gallery generated by Sphinx-Gallery