Create a discipline from an external executable

from __future__ import annotations

import os
import subprocess

from numpy import array

from gemseo import configure_logger
from gemseo.core.discipline import MDODiscipline

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 MDODiscipline consists in three steps:

1. Instantiate the MDODiscipline using the super constructor,

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

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

The MDODiscipline._run method consists in three steps:

1. Get the input data from MDODiscipline.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 MDODiscipline.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(MDODiscipline):
    def __init__(self) -> None:
        super().__init__("ShellDisc")
        self.input_grammar.update_from_names(["a", "b"])
        self.output_grammar.update_from_names(["c"])
        self.default_inputs = {"a": array([1.0]), "b": array([2.0])}

    def _run(self) -> None:
        cwd = os.getcwd()
        inputs_file = os.path.join(cwd, "inputs.txt")
        outputs_file = os.path.join(cwd, "outputs.txt")
        write_file(self.local_data, inputs_file)
        subprocess.run("python run.py".split(), cwd=cwd)
        outputs = parse_file(outputs_file)
        self.local_data.update(outputs)

Execution of the discipline

Now we can run it with default input values:

shell_disc = ShellExecutableDiscipline()
shell_disc.execute()

{'b': array([2.]), 'a': array([1.]), '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.])}