Common Python Interface of Applications for Users

Overview

Common Interface of Applications: The AnalysisStage

Overview of the AnalysisStage

This page describes a common set of functions to be implemented by the AnalysisStage class of each application to make it possible to use the application as an object. This facilitates to combine applications in a single analysis, in a modular way, to do multi-physics and/or co-simulation.

Each application implements a Main Analysis Script (formerly implemented in the MainKratos.py). This script can then be used either for coupling applications or running regular simulation of an individual application.

The AnalysisStage object of each application can be constructed (after importing the script) with ProjectParameters (a Kratos::Parameters object) or by passing the ProjectParameters JSON file.

Once the object of the AnalysisStage for an application is constructed, the simulation can be run using a set of interface functions. These functions are aimed to provide different levels of control over a simulation.

Interface Functions for running a simulation using Analysis object of an application:

The following set of functions are aimed to provide a fine control over the simulation.

Initialize Initializes the analysis (called once at the beginning of the simulation)
InitializeTimeStep Initializes the Timestep (called once at the beginning of a timestep)
SolveTimeStep Solves the time step (can be called several times within a timestep (e.g. strong coupling in FSI) )
FinalizeTimeStep Finalizes the Timestep (called once at the end of a timestep)
Finalize Finalizes the analysis (called once at the end of the simulation)

In addition to the above mentioned functions, two more functions which allow to execute the whole simulation with less control over the simulation.

Run executes the entire analysis
RunMainTemporalLoop Executes the entire time loop

Base Class of the AnalysisStage Object (in Kratos Core):

from __future__ import print_function, absolute_import, division  # makes KratosMultiphysics backward compatible with python 2.6 and 2.7

# Importing Kratos
import KratosMultiphysics

class BaseKratosAnalysisStage(object):
    """The base class for the analysis classes in the applications
    """
    def __init__(self, project_parameters, external_model_part=None):
        """The constructor of the Analysis-Object.
        It obtains the project parameters used for the analysis
        This function is intended to be called from the constructor
        of deriving classes:
        super(DerivedAnalysis, self).__init__(project_parameters)

        It is intended that this constructor creates the solver and
        adds the Variables that are needed by the solver to the
        ModelPart

        An external ModelPart is used if the AnalysisStage is used
        in a larger context, e.g. in a MultiStage-Analysis or Optimization
        """
        if (type(project_parameters) == str): # a file name is provided
            with open(project_parameters,'r') as parameter_file:
                self.ProjectParameters = KratosMultiphysics.Parameters(parameter_file.read())
        elif (type(project_parameters) == KratosMultiphysics.Parameters): # a Parameters object is provided
            self.ProjectParameters = project_parameters
        else:
            raise Exception("Input is expected to be provided as a Kratos Parameters object or a file name")

        if external_model_part is not None:
            if (type(external_model_part) != KratosMultiphysics.ModelPart):
                raise Exception("Input is expected to be provided as a Kratos ModelPart object")
            self.using_external_model_part = True

        ## Get echo level and parallel type
        self.echo_level = self.ProjectParameters["problem_data"]["echo_level"].GetInt()
        self.parallel_type = self.ProjectParameters["problem_data"]["parallel_type"].GetString()

        ## Import parallel modules if needed
        if (self.parallel_type == "MPI"):
            import KratosMultiphysics.mpi as KratosMPI
            import KratosMultiphysics.MetisApplication as MetisApplication
            import KratosMultiphysics.TrilinosApplication as TrilinosApplication
            self.is_printing_rank = (KratosMPI.mpi.rank == 0)
        else:
            self.is_printing_rank = True

        ## model part definition
        if self.using_external_model_part:
            self.main_model_part = external_model_part
        else:
            main_model_part_name = self.ProjectParameters["problem_data"]["model_part_name"].GetString()
            self.main_model_part = KratosMultiphysics.ModelPart(main_model_part_name)
            self.main_model_part.ProcessInfo.SetValue(KratosMultiphysics.DOMAIN_SIZE,
                                                      self.ProjectParameters["problem_data"]["domain_size"].GetInt())

        self._CreateSolver()

        self.solver.AddVariables()

        if not self.using_external_model_part:
            ## Read the model
            self._ReadModelPart()


    #### Public functions to run the Analysis ####
    def Run(self):
        """This function executes the entire analysis
        It is NOT intended to be overridden in deriving classes!
        """
        self.Initialize()
        self.RunMainTemporalLoop()
        self.Finalize()

    def RunMainTemporalLoop(self):
        """This function executes the temporal loop of the analysis
        It is NOT intended to be overridden in deriving classes!
        """
        while self.time < self.end_time:
            self.InitializeTimeStep()
            self.SolveTimeStep()
            self.FinalizeTimeStep()

    def Initialize(self):
        """This function initializes the analysis
        Usage: It is designed to be called ONCE, BEFORE the execution of the time-loop
        This function IS intended to be overridden in deriving classes!
        At the end of this function the StageAnalysis is ready for the time-loop
        It should be called AFTER the ModelPart used for this AnalysisStage is used
        """
        pass

    def Finalize(self):
        """This function finalizes the analysis
        Usage: It is designed to be called ONCE, AFTER the execution of the time-loop
        This function IS intended to be overridden in deriving classes!
        """
        pass

    def InitializeTimeStep(self):
        """This function initializes the time-step
        Usage: It is designed to be called once at the beginning of EACH time-step
        This function IS intended to be overridden in deriving classes!
        """
        pass

    def SolveStep(self):
        """This function solves one step
        It can be called several times during one time-step
        This is equivalent to calling "solving_strategy.Solve()" (without "Initialize")
        This function is NOT intended to be overridden in deriving classes!
        """
        self.InitializeSolutionStep()
        self.Predict()
        self.SolveSolutionStep()
        self.FinalizeSolutionStep()

    def FinalizeTimeStep(self):
        """This function finalizes the time-step
        Usage: It is designed to be called once at the end of EACH time-step
        This function IS intended to be overridden in deriving classes!
        """
        pass

    def InitializeSolutionStep(self):
        """This function performs all the required operations that should be done
        (for each step) before solving the solution step.
        This function has to be implemented in deriving classes!
        """
        raise NotImplementedError("This function has to be implemented by derived\
            analysis classes")

    def Predict(self):
        """This function predicts the solution
        This function has to be implemented in deriving classes!
        """
        raise NotImplementedError("This function has to be implemented by derived\
            analysis classes")

    def SolveSolutionStep(self):
        """This function solves the current step
        This function has to be implemented in deriving classes!
        """
        raise NotImplementedError("This function has to be implemented by derived\
            analysis classes")

    def FinalizeSolutionStep(self):
        """This function Performs all the required operations that should be done
        (for each step) after solving the solution step.
        This function has to be implemented in deriving classes!
        """
        raise NotImplementedError("This function has to be implemented by derived\
            analysis classes")

    def _ReadModelPart(self):
        """This function reads the ModelPart, in case it is not provided to the AnalysisStage
        This function is NOT intended to be overridden in deriving classes!
        """
        if(self.settings["model_import_settings"]["input_type"].GetString() == "mdpa"):
            # Import model part from mdpa file.
            KratosMultiphysics.Logger.PrintInfo("::[MechanicalSolver]::", "Reading model part from file: " + os.path.join(problem_path, input_filename) + ".mdpa")
            KratosMultiphysics.ModelPartIO(input_filename).ReadModelPart(self.main_model_part)
            KratosMultiphysics.Logger.PrintInfo("::[MechanicalSolver]::", "Finished reading model part from mdpa file.")
        elif
            ...
        else:
            Error
        # import materials if applicable
        if self.ProjectParameters["material_import_settings"].Has("materials_filename"):
            materials_filename = self.ProjectParameters["material_import_settings"]["materials_filename"].GetString()
            if (materials_filename != ""):
                import read_materials_process
                # Create a dictionary of model parts.
                Model = KratosMultiphysics.Model()
                Model.AddModelPart(self.main_model_part)
                # Add constitutive laws and material properties from json file to model parts.
                read_materials_process.ReadMaterialsProcess(Model, self.settings["material_import_settings"])

    def _CreateSolver(self):
        """This function creates the solver of the AnalysisStage
        (typically by importing it using the python solver wrappers)
        This function has to be implemented in deriving classes!
        """
        raise NotImplementedError("This function has to be implemented by derived\
            analysis classes")

The naming convention of this Main Analysis Script is the following: application_name_in_lower_case_analysis.py

e.g. structural_mechanics_analysis.py (for the StructuralMechanicsApplication) or fluid_dynamics_analysis.py (for the FluidDynamicsApplication)

The name of the main class is specified as ApplicationNameAnalysis, e.g. StructuralMechanicsAnalysis, which can be constructed with a Kratos::Parameters object or a project parameters JSON file.

Running an individual full simulation using the Main Analysis Script

The Main Analysis Script script can be also made callable, to perform an individual analysis can be performed by calling python on it.

The following example shows how it can be done (from: StructuralMechanicsApplication)

if __name__ == "__main__":
    from sys import argv

    if len(argv) > 2:
        err_msg =  'Too many input arguments!\n'
        err_msg += 'Use this script in the following way:\n'
        err_msg += '- With default ProjectParameters (read from "ProjectParameters.json"):\n'
        err_msg += '    "python3 structural_mechanics_analysis.py"\n'
        err_msg += '- With custom ProjectParameters:\n'
        err_msg += '    "python3 structural_mechanics_analysis.py CustomProjectParameters.json"\n'
        raise Exception(err_msg)

    if len(argv) == 2: # ProjectParameters is being passed from outside
        project_parameters_file_name = argv[1]
    else: # using default name
        project_parameters_file_name = "ProjectParameters.json"

    StructuralMechanicsAnalysis(project_parameters_file_name).Run()

Using Main Analysis Script in Co-Simulation or Multi-Physics simulation

One of the advantages of having the Main Analysis Script with the Analysis class with the interface functions defined above is that, it makes the usage of two or more Analyses together in a co-simulation. The following code snippet illustrates the same using StructuralMechanics and FluidDynamics analysis objects.

from KratosMultiphysics import *
from structural_mechanics_analysis import *
from fluid_dynamics_analysis import *

structural_analysis = StructuralMechanicsAnalysis(StructureProjectParameters.json)
fluid_analysis = FluidDynamicsAnalysis(FluidProjectParameters.json)

structural_analysis.Initialize()
fluid_analysis.Initialize()

while (time<time_end):
  structural_analysis.InitializeTimeStep()
  fluid_analysis.InitializeTimeStep()
  
  # Transfer data from structural_analysis to fluid_analysis

  fluid_analysis.SolveTimeStep()

  # Transfer data from fluid_analysis to structural_analysis

  structural_analysis.SolveTimeStep()

  structural_analysis.FinalizeTimeStep()
  fluid_analysis.FinalizeTimeStep()  



structural_analysis.Finalize()
fluid_analysis.Finalize()

Note that this type of coupling is done if several domains are involved (e.g. FSI) => this can be called "AnalysisStageCoupling"

For coupling on the same domain (i.e. where e.g. Nodes are shared, done in ALE-Fluid or ThermoMechanical), it is recommended to do the coupling on the Solver-Level, e.g. due to technical complications, e.g. the CloneTimeStep-issue

Note for developers

It is recommended to embed this script in the application tests in order to ensure that is is working properly

Python-Solvers

The AnalysisStage Object is designed to solve an entire Simulation. This means that it sets up the Boundary conditions, IO, reading of the ModelPart and Materials, controlling the time-stepping etc. It does not know abt how to solve the physical problem. This is the task of the Solver. This class implements everything that is related to solve the physical problem, e.g. it constructs Strategy, BuilderAndSolver,... Everything else (e.g. reading the ModelPart) is excluded form the Solver and part of the AnalysisStage.

Furthermore, it knows abt which Variables and DOFs it needs and adds them to the ModelPart It does NOT read the ModelPart, this is task of the AnalysisStage. After the ModelPart is read, the solver might need to perform some operations on it (e.g. create the ComputingModelPart), this is done with PrepareModelPartForSolver.

The idea is that the AnalysisStage has a Solver as a member. This way the responsibilities of each object are separated and no BLOB is created.

A draft implementation for the BaseSolver is shown in the following (following what is currently done in FLuidDynamics and StructuralMechanics):

Base Class of the Solver Object (in Kratos Core):

from __future__ import print_function, absolute_import, division  # makes KratosMultiphysics backward compatible with python 2.6 and 2.7

# Importing Kratos
import KratosMultiphysics

class BaseKratosSolver(object):
    """The base class for the python solvers in the applications
    This class knows abt the solution of the physical problem
    """
    def __init__(self, main_model_part, project_parameters):
        """The constructor of the Solver.
        It obtains the project parameters
        This function is intended to be called from the constructor
        of deriving classes:
        super(DerivedSolver, self).__init__(project_parameters)
        """
        if (type(main_model_part) != KratosMultiphysics.ModelPart):
            raise Exception("Input is expected to be provided as a Kratos ModelPart object")

        if (type(custom_settings) != KratosMultiphysics.Parameters):
            raise Exception("Input is expected to be provided as a Kratos Parameters object")

        self.main_model_part = main_model_part
        self.project_parameters = project_parameters

        self._ValidateParameters()

    #### Public functions to run the Analysis ####
    def AddVariables(self):
        pass
    def AddDofs(self):
        pass
    def GetMinimumBufferSize(self):
        pass
    def PrepareModelPartForSolver(self):
        pass
    def _ValidateParameters(self):
        pass
    def ComputeDeltaTime(self):
        pass
    def Inizialize(self):
        pass
    def GetComputingModelPart(self):
        pass
    def SetEchoLevel(self):
        pass
    def Clear(self):
        pass
    def Check(self):
        pass

    def Solve(self):
        """This function solves one step
        It can be called several times during one time-step
        This is equivalent to calling "solving_strategy.Solve()" (without "Initialize")
        This function is NOT intended to be overridden in deriving classes!
        """
        self.InitializeSolutionStep()
        self.Predict()
        self.SolveSolutionStep()
        self.FinalizeSolutionStep()

    def InitializeSolutionStep(self):
        """This function performs all the required operations that should be done
        (for each step) before solving the solution step.
        This function has to be implemented in deriving classes!
        """
        raise NotImplementedError("This function has to be implemented by derived\
            analysis classes")

    def Predict(self):
        """This function predicts the solution
        This function has to be implemented in deriving classes!
        """
        raise NotImplementedError("This function has to be implemented by derived\
            analysis classes")

    def SolveSolutionStep(self):
        """This function solves the current step
        This function has to be implemented in deriving classes!
        """
        raise NotImplementedError("This function has to be implemented by derived\
            analysis classes")

    def FinalizeSolutionStep(self):
        """This function Performs all the required operations that should be done
        (for each step) after solving the solution step.
        This function has to be implemented in deriving classes!
        """
        raise NotImplementedError("This function has to be implemented by derived\
            analysis classes")

Outlook (Kratos-Project, Multi-Stage Simulation)

Note This is a collection of ideas, to be done AFTER AnalysisStage and Solver are implemented in a first version. Please note that the following is in a very early design phase.

In the future the objects presented here can be used in a larger context, e.g. a Multi-Stage Analysis. This means that e.g. a FormFinding Analysis can be performed with doing a FSI-simulation afterwards. The above mentioned objects are already designed for this, e.g. a ModelPart can be passed from outside to the AnalysisStage, this means that it can be used in severals AnalysisStages.

The idea is that in the beginning all AnalysisStages are constructed (i.e. all necessary Variables are added to the ModelPart), then the ModelPart is being read. This can be done e.g. by a global ModelManager. For this to work the Model has to be enhanced, therefore it should be done later.

This could look like this:

import KratosMultiphysics

##construct all the stages  

Model = KratosMultiphysics.Kernel().GetModel() #if we want it to be somewhere else more than fine
list_of_analysis_stages = GenerateStages(Model, "ProjectParameters.json") #internally loads the applications needed

model_manager.Read(list_of_analysis_stages, Model)
for solver in list_of_solvers:
    solver.Initialize()
    solver.Run()
    solver.Finalize()

Project information

Getting Started

Getting Kratos (Last compiled Release)
- Kratos for Linux
- Kratos for Windows
- Kratos for GiD
Compiling Kratos
- See INSTALL.md

Tutorials

Developers

Kratos structure

Conventions

Local Axes Orientation

Solvers

Debugging, profiling and testing

HOW TOs

Utilities

Kratos API

MultiPoint Constraint (MPC) interface

Kratos Structural Mechanics API

Documentation about the Constitutive Laws

Provide feedback

Saved searches

Use saved searches to filter your results more quickly