Pre run post model

pre_run_post_model

Classes

PreRunPostModel

PreRunPostModel(load_case_name: str, **options)

Bases: IntegratedModel

A model chaining a pre-processor, a run-processor, and a post-processor.

An PreRunPostModel contains three :class:~.ExternalSoftwareComponent: The model execution is basically the sequenced execution of these three chained components.

The description of the components is done through the following class attributes: * PRE_PROC_FAMILY * RUN_FAMILY * POST_PROC_FAMILY For instance, PRE_PROC_FAMILY = "PreStraightBeam" corresponds to the preprocessor family of the analytical beam model StraightBeamModel. If this model is instantiated for the Cantilever load case, the model preprocessor class is PreStraightBeam_Cantilever according to the naming convention {PRE_PROC_FAMILY}_{load_case} (and similarly for the postprocessor).

For FE models using a user subroutine, the list of subroutines used is defined in class attribute SUBROUTINE_NAMES.

Notes

Since an IntegratedModel extends a GEMSEO <https://gemseo.readthedocs.io/en/stable/index.html> :class:~.gemseo.core.discipline.MDODiscipline, it can be used as |gemseo| discipline <https://gemseo.readthedocs.io/en/6.0.0/examples/api /plot_discipline.html#discipline>. Also, |gemseo| capabilities (MLearning, Optimisation) can be used on the model.

Examples:

>>> # Run a model with default values, each execution being stored in a unique
>>> # directory. The archive and scratch persistency is customized.
>>> # The manager for archiving the results is set to MLFlow.
>>> from vimseo.api import create_model
>>> from vimseo.core.model_settings import IntegratedModelSettings
>>> model = create_model("BendingTestAnalytical", "Cantilever",
>>>     model_options=IntegratedModelSettings(
>>>         directory_archive_root=f"my_model_result",
>>>         directory_scratch_root=f"my_scratch",
>>>         directory_scratch_persistency=PersistencyPolicy.DELETE_ALWAYS,
>>>         directory_archive_persistency=PersistencyPolicy.DELETE_IF_FAULTY,
>>>         archive_manager="MlflowArchive",
>>>     ),
>>> )
>>> model.execute()

Initialize self. See help(type(self)) for accurate signature.

Parameters:

• load_case_name (str) –

  The description is missing.

• **options –

  The description is missing.

Source code in src/vimseo/core/pre_run_post_model.py

def __init__(self, load_case_name: str, **options):

    options = IntegratedModelSettings(**options).model_dump()
    material = (
        Material.from_json(self.MATERIAL_FILE) if self.MATERIAL_FILE != "" else None
    )

    component_factory = ComponentFactory()

    run_processor = component_factory.create(
        self.RUN_FAMILY,
        material_grammar_file=self._MATERIAL_GRAMMAR_FILE,
        material=material,
        check_subprocess=options["check_subprocess"],
    )
    self._subroutine_names = deepcopy(self.SUBROUTINES_NAMES)
    for sub_name in self._subroutine_names:
        run_processor.subroutine_list.append(
            SubroutineWrapperFactory().create(sub_name)
        )

    load_case = LoadCaseFactory().create(
        load_case_name, domain=self._LOAD_CASE_DOMAIN
    )
    components = [
        component_factory.create(
            self.PRE_PROC_FAMILY,
            load_case_name,
            load_case=load_case,
            material_grammar_file=self._MATERIAL_GRAMMAR_FILE,
            material=material,
            check_subprocess=options["check_subprocess"],
        ),
        run_processor,
        component_factory.create(
            self.POST_PROC_FAMILY,
            load_case_name,
            load_case=load_case,
            check_subprocess=options["check_subprocess"],
        ),
    ]

    # run component has its grammar defined from pre and post components:
    components[1].input_grammar = deepcopy(components[0].output_grammar)
    components[1].input_grammar.update(components[0].input_grammar)
    components[1].output_grammar = deepcopy(components[2].input_grammar)

    super().__init__(load_case_name, components, **options)

    self._pre_processor = self._chain.disciplines[0]
    self._run_processor = self._chain.disciplines[1]
    self._post_processor = self._chain.disciplines[2]

    self._component_with_jacobian = self.run
    self._set_differentiated_names(self.run)

    self.run.job_executor._user_job_options.update({"n_cpus": self.N_CPUS})

Attributes

CURVES class-attribute

CURVES: Sequence[tuple[str]] = []

The output data to plot as curves. Define a tuple for each curve. It can be load case independent or dependent. Only the first two elements of the list of variables are considered.The first variable is the abscissa variable and the second value is the ordinate.

Example

CURVES = [('strain_history', 'stress_history')]

MATERIAL_FILE class-attribute

MATERIAL_FILE: Path | str = ''

The path to the json file defining the material values.

NUMERICAL_VARIABLE_NAMES class-attribute

NUMERICAL_VARIABLE_NAMES: Sequence[str] = []

The input variable names which are numerical parameters.

N_CPUS class-attribute instance-attribute

N_CPUS = 1

The default number of cpus used to run the model.

POST_PROC_FAMILY class-attribute instance-attribute

POST_PROC_FAMILY = None

The prefix of the post-processor class name.

PRE_PROC_FAMILY class-attribute instance-attribute

PRE_PROC_FAMILY = None

The prefix of the pre-processor class name (typically MyPre for a pre- processor class named MyPre_MyLoadCase).

RUN_FAMILY class-attribute instance-attribute

RUN_FAMILY = None

The prefix of the run-processor class name.

SUBROUTINES_NAMES class-attribute

SUBROUTINES_NAMES: Sequence[str] = []

The names of the subroutines.

SUMMARY class-attribute

SUMMARY: str = ''

A brief description of the model (purpose, range of application, hypothesis, limitations, similar models).

archive_manager property

archive_manager

The archive manager.

boundary_condition_variable_names property

boundary_condition_variable_names

The input variable names which are boundary condition parameters.

description property

description

The model description.

geometrical_variable_names property

geometrical_variable_names

The input variable names which are geometrical parameters.

image_path property

image_path: Path | None

The fully-qualified path to the image illustrating the model.

If no image is associated to the model, the path to the image associated with the load case is returned.

job_description class-attribute instance-attribute

job_description: str = ''

Free string description of the job to be executed, to be passed as a metadata result.

load_case property

load_case: LoadCase

The load case.

material property

material: Material

The material.

material_variable_names property

material_variable_names

The input variable names which are material parameters.

n_cpus property

n_cpus

The number of CPUs to run this model.

numerical_variable_names property

numerical_variable_names

The input variable names which are numerical parameters.

run property

run: RunProcessor

The component running the external software.

scratch_job_directory property

scratch_job_directory: Path | str

The Path to the scratch job directory, once the model is executed.

Before model execution, or if the model uses the cache, the empty string is returned.

Classes

InputGroupNames

Bases: StrEnum

The names of the input variable groups.

Functions

auto_get_file

auto_get_file(
    file_suffix: str, load_cases: Iterable[str] = ()
) -> Iterable[Path]

Return file paths for files following a naming convention and placed next to the model class or its parents, or next to a model load case class. Naming convention is {model_class_name}{file_suffix} or {model_class_name}_{load_case}{file_suffix}.

Parameters:

• file_suffix (str) –

  The suffix, including the extension, of the file to be searched in next to model class or its parents.

• load_cases (Iterable[str], default: () ) –

  an iterable of load cases. If specified, only the files placed next to these load case classes are searched. Otherwise, all load case classes associated to the model are considered.

Returns:

• Iterable[Path] –

  A list of file paths.

Source code in src/vimseo/core/base_integrated_model.py

def auto_get_file(
    self,
    file_suffix: str,
    load_cases: Iterable[str] = (),
) -> Iterable[Path]:
    """Return file paths for files following a naming convention and placed next to
    the model class or its parents, or next to a model load case class. Naming
    convention is {model_class_name}{file_suffix} or
    {model_class_name}_{load_case}{file_suffix}.

    Args:
        file_suffix: The suffix, including the extension, of the file to be searched
            in next to model class or its parents.
        load_cases: an iterable of load cases. If specified, only the files placed
            next to these load case classes are searched.
            Otherwise, all load case classes associated to the model are considered.

    Returns:
        A list of file paths.
    """
    from vimseo.api import get_available_load_cases

    cls = self.__class__
    model_name = cls.__name__
    load_cases = load_cases or get_available_load_cases(model_name)
    classes = [cls] + [
        base for base in cls.__bases__ if issubclass(base, Discipline)
    ]
    names = [model_name] + [cls.__name__ for cls in classes[1:]]

    matching_files = []
    for cls, name in zip(classes, names, strict=False):
        class_module = sys.modules[cls.__module__]
        directory_path = Path(class_module.__file__).parent.absolute()
        file_path = directory_path / f"{name}{file_suffix}"
        if file_path.is_file():
            matching_files.append(file_path)
        for load_case in load_cases:
            file_path = directory_path / f"{name}_{load_case}{file_suffix}"
            if file_path.is_file():
                matching_files.append(file_path)

    if not matching_files:
        msg = (
            f"No files with suffix {file_suffix} were found next "
            f"to the model {self.name} or its parent class."
        )
        raise FileNotFoundError(msg)

    return matching_files

create_cache_from_archive

create_cache_from_archive(
    run_ids: Iterable[str] = (),
) -> HDF5Cache

Defines a temporary HDF5 cache file, based on results found on the current archive.

Source code in src/vimseo/core/base_integrated_model.py

def create_cache_from_archive(self, run_ids: Iterable[str] = ()) -> HDF5Cache:
    """Defines a temporary HDF5 cache file, based on results found on the current
    archive."""

    # cache file preparation
    cache_dir_path = Path(self._cache_file_path).parent
    self._cache_file_path = Path(
        cache_dir_path
        / f"{self.__class__.__name__}_{self.__load_case.name}_from_archive.hdf"
    )
    self._cache_file_path.unlink(
        missing_ok=True
    )  # delete possible old tmp cache file
    # note that tolerance is important for now: see issue #233.

    self.set_cache(
        cache_type=Discipline.CacheType.HDF5,
        hdf_file_path=self._cache_file_path,
        hdf_node_path="node",
        tolerance=1.0e-10,
    )

    results = self._archive_manager.get_archived_results(run_ids=run_ids)
    if results is None:
        return None

    for result in results:
        if not isinstance(self.cache, SimpleCache):
            to_array = self.output_grammar.data_converter.convert_value_to_array
            for name, value in result["outputs"].items():
                result["outputs"][name] = to_array(name, value)

        self.cache.cache_outputs(result["inputs"], result["outputs"])
        # jacobian = None
        #
        # # append a cache entry with the current result
        # self.cache[result["inputs"]] = (result["outputs"], jacobian)

    return self.cache

execute

execute(
    input_data: StrKeyMapping = READ_ONLY_EMPTY_DICT,
) -> DisciplineData

Execute the discipline, i.e. compute output data from input data.

If :attr:.virtual_execution is True, this method returns the :attr:.default_output_data. Otherwise, it calls the :meth:._run method performing the true execution and returns the corresponding output data. This :meth:._run method must be implemented in subclasses.

Parameters:

• input_data (StrKeyMapping, default: READ_ONLY_EMPTY_DICT ) –

  The input data. Complete this dictionary with the :attr:.default_input_data.

Returns:

• DisciplineData –

  The input and output data.

Source code in src/vimseo/core/gemseo_discipline_wrapper.py

def execute(
    self,
    input_data: StrKeyMapping = READ_ONLY_EMPTY_DICT,
) -> DisciplineData:

    if self.EXTRA_INPUT_GRAMMAR_CHECK:
        from vimseo.core.base_integrated_model import IntegratedModel
        from vimseo.core.components.pre.pre_processor import PreProcessor

        all_input_names = list(input_data.keys()) + list(
            self.default_input_data.keys()
        )

        if (isinstance(self, (IntegratedModel, PreProcessor))) and not set(
            all_input_names
        ).issubset(set(self.input_grammar.names)):
            extra_inputs = list(
                set(all_input_names) - set(self.input_grammar.names)
            )
            msg = (
                f"Input {extra_inputs} are not defined in the input grammar."
                f"Input grammar names are {list(self.input_grammar.names)}."
            )
            raise KeyError(msg)

    return super().execute(input_data)

get_dataflow

get_dataflow()

The inputs and outputs of the components of the model.

Returns:

• –

  A dictionary mapping component name to its input and output names.

Source code in src/vimseo/core/pre_run_post_model.py

def get_dataflow(self):
    """The inputs and outputs of the components of the model.

    Returns:
         A dictionary mapping component name to its input and output names.
    """

    dataflow = super().get_dataflow()
    dataflow["subroutine_names"] = self._subroutine_names

    return dataflow

get_output_data_names

get_output_data_names(
    remove_metadata: bool = False,
) -> list[str]

Parameters:

• remove_metadata (bool, default: False ) –

  Whether to remove the metadata from the output data names.

Source code in src/vimseo/core/base_integrated_model.py

def get_output_data_names(self, remove_metadata: bool = False) -> list[str]:
    """

    Args:
        remove_metadata: Whether to remove the metadata from the output data names.

    """
    output_data_names = list(self.output_grammar.names)

    if remove_metadata:
        for name in self.get_metadata_names():
            if name in output_data_names:
                output_data_names.remove(name)

    return output_data_names

plot_results

plot_results(
    directory_path: str | Path = "",
    save: bool = True,
    show: bool = False,
    scalar_names: Iterable[str] = (),
    data: str = "CURVES",
) -> Mapping[str, Figure]

Plots the results as curves, from definition of curves in CURVES variable.

Parameters:

• directory_path (str | Path, default: '' ) –

  A path where to save the plots. Default is current working directory.

• save (bool, default: True ) –

  Whether to save the plot.

• show (bool, default: False ) –

  Whether to show the plot.

• scalar_names (Iterable[str], default: () ) –

  The scalars to plot in a scatter matrix. Show all scalars by default.

Returns:

Source code in src/vimseo/core/base_integrated_model.py

def plot_results(
    self,
    directory_path: str | Path = "",
    save: bool = True,
    show: bool = False,
    scalar_names: Iterable[str] = (),
    data: str = "CURVES",
) -> Mapping[str, Figure]:
    """Plots the results as curves, from definition of curves in CURVES variable.

    Args:
        directory_path: A path where to save the plots. Default is current working
            directory.
        save: Whether to save the plot.
        show: Whether to show the plot.
        scalar_names: The scalars to plot in a scatter matrix. Show all scalars by default.

    Returns:
    """
    directory_path = Path.cwd() if directory_path == "" else Path(directory_path)
    if not directory_path.exists():
        directory_path.mkdir(parents=True)

    LOGGER.info(f"Saving plots to {directory_path.absolute()}")

    from vimseo.core.model_result import ModelResult

    result = ModelResult.from_data(
        {"inputs": self.get_input_data(), "outputs": self.get_output_data()},
        model=self,
    )
    figures = {}
    if data == "CURVES":
        for variables in self.curves:
            file_name = (
                f"{self.name}_{self.__load_case.name}_"
                f"{variables[1]}_vs_{variables[0]}.html"
            )
            figures[f"{variables[1]}_vs_{variables[0]}"] = plot_curves(
                result.get_curve(variables),
                directory_path=directory_path,
                file_name=file_name,
                save=save,
                show=show,
            )
            LOGGER.info(
                f"Plot {file_name} is saved to {Path(directory_path) / file_name}"
            )
    elif data == "SCALARS":
        plot = ScatterMatrix(
            Dataset.from_dataframe(
                DataFrame.from_dict([
                    result.get_numeric_scalars(
                        variable_names=scalar_names,
                    )
                ])
            )
        )
        figures["scalars"] = plot.execute(
            directory_path=directory_path,
            file_name="scalars.html",
            save=False,
            show=True,
        )
    else:
        msg = f"Unknown data type {data} for plotting."
        raise ValueError(msg)

    return figures

reset_cache

reset_cache(new_cache_path: str | Path)

Create a new cache with another file path.

Has an effect only for HDF5 cache.

Source code in src/vimseo/core/base_integrated_model.py

def reset_cache(self, new_cache_path: str | Path):
    """Create a new cache with another file path.

    Has an effect only for ``HDF5`` cache.
    """
    self._cache_file_path = new_cache_path
    self.set_cache(
        cache_type=CacheType.HDF5,
        hdf_file_path=self._cache_file_path,
        hdf_node_path="node",
    )

show_image

show_image() -> None

Show the image illustrating the load case.

Source code in src/vimseo/core/base_integrated_model.py

def show_image(self) -> None:
    """Show the image illustrating the load case."""
    if self.image_path:
        imshow(asarray(imread(self.image_path)))