TransAT API documentation¶

TransAT API is composed of

a user configuration file (user_config.ini)
a main Simulation class, exposing high-level methods
an app builder, for fast prototyping of python applications
several specialized modules, exposing low-level methods

TransAT API is currently implemented in Python. Interfaces with other languagues are available upon request.

TransAT python package documentation

Installation¶

A step-by-step installation guide can be found a here.

Configuration¶

TransAT-API acts as a glue between all the the components involved in CFD simulation process (CAD - Solver - Post processing). Configurations such as paths and working directory are defined in the file user_config.ini. User can generate a template file by calling

python -m transat.config.create_user_config

Then open this file and edit it.

Name	Description
wd	working directory, where the files will be saved.
tmb_path	path to the bin folder of transatMB
path: ui	path to the bin folder of transatUI
path: freecad	path to the lib folder of freecad

Examples¶

These are short examples of how the API can be used.

2D Pipe (server example)¶

The following example simulate a steady single phase flow in a 2D pipe. The simulation is launched on TransAT-Server. First a server is initialized with the function launch_server. It returns a URL at which the server can be fetched. This function could be launched outside my_func and the URL can be passed to the app by the arguments.

url = launch_server()

The client will continuously fetch the server to get the current iteration (watch function). This function print the current iteration every second.

This example can be run by typing the following command:

python  -m transat.apps.pipe.pipe

from transat.simulation import Simulation
import transat.appbuilder as appbuilder
from transat.server.server import Server
import threading
from transat.server.client import Client
import time
import os

app = appbuilder.App('Pipe')


def launch_server():
    se = Server()
    server = threading.Thread(target=se.run)
    server.daemon = True
    server.start()
    return se.get_address()


def get_postprocess_path(name):
    path = os.path.realpath(__file__)
    path = os.path.dirname(path)
    return os.path.join(path, name)


def watch(url):
    cl = Client(url)
    iteration = None
    while cl.has_jobs():
        _iteration = cl.get_current_iteration()
        if iteration != _iteration:
            iteration = _iteration
            print "iteration " + str(iteration)
        time.sleep(1)

    print cl.get_current_timestep()


def get_center_line(sim):
    y = sim.setup.mesher.get_center('y')
    z = sim.setup.mesher.get_center('z')
    x1 = sim.setup.mesher.get_min('x')
    x2 = sim.setup.mesher.get_max('x')
    return [[x1, y, z], [x2, y, z]]


def my_fun(args):
    # Create a TransAT server
    url = launch_server()

    sim = Simulation("Pipe")
    # simulations is saved in {local_wd}/Pipe/test_pipe
    sim.set_folder('test_pipe', ['local'])

    # load templates. It will first look in the relative directory
    # and then in the installation directory
    sim.load_template('templates/pipe')

    # create mesh and files needed by the TransAT solver
    sim.prepare_simulation_files()

    sim.run_init(worker='local', nprocs=3, url=url)
    sim.run(worker='local', nprocs=3, url=url)

    watch(url)

    center_line = get_center_line(sim)

    sim.add_postprocess(get_postprocess_path("pipe_postprocess.py"))
    sim.run_postprocess('plot_pressure',
                        worker_name='local',
                        args={'points': center_line,
                              'folder': sim.load_path('local')})


app.add_main(my_fun)

if __name__ == '__main__':
    args = app.parse_inputs()
    app.run(args)

The output of this app a figure with the pressure drop along the pipe.

3D Pipe¶

The radius of the pipe is parametrized and is given as argument to the application.

python  -m transat.apps.3D_pipe.pipe --radius=0.5

from transat.simulation import Simulation
import transat.appbuilder as appbuilder
from transat.server.server import Server
import threading
from transat.server.client import Client
import time
import os
from transat.setup.cad import Pipe
from transat.setup.cad import Network
from transat.config import ascomp_setup as setup
from transat.setup.cad import CAD
global_config = setup.install()

app = appbuilder.App('3D_Pipe')
app.add_input('radius', unit='m')


def create_pipe(pipe_radius = 0.25):
    cad = CAD(freecad_path=global_config.env['path']['freecad'])

    pipe = Pipe(name="pipe.stl", radius=pipe_radius)
    pipe.add_point([0, 0.0, 0])
    pipe.add_point([10, 0.0, 0])

    network = Network(cad, [pipe])
    return network, pipe

def get_postprocess_path(name):
    path = os.path.realpath(__file__)
    path = os.path.dirname(path)
    return os.path.join(path, name)



def my_fun(args):

    sim = Simulation("3D_Pipe")
    # simulations is saved in {local_wd}/Pipe/test_pipe
    sim.set_folder('test_pipe', ['local'])

    # load templates. It will first look in the relative directory
    # and then in the installation directory
    sim.load_template('templates/network_single')

    radius = float(args['radius'])

    network, pipe = create_pipe(radius)
    network.create(path=sim.load_path('local'))

    print sim.load_path('local')

    sim.setup.mesher.set_blocks(network.get_blocks())
    sim.setup.mesher.split_blocks()
    sim.setup.ist.add(network.get_pipes_name())

    sim.setup.mesher.set_dx(radius / 5.0)

    sim.setup.bcs.get_bc('inflow').set_velocities([0.001, 0, 0])

    tree = sim.setup.bcs.create_surfaces_tree()
    tree.get_all_surfaces().set_bc_name('wall')

    tree.get_surface_with_point(pipe.get_inlet_point()).set_bc_name('inflow')
    tree.get_surface_with_point(pipe.get_outlet_point()).set_bc_name('outflow')

    sim.prepare_simulation_files()


    sim.run_init(worker='local', nprocs=2)
    sim.run(worker='local', nprocs=2)


    y = sim.setup.mesher.get_center('y')
    z = sim.setup.mesher.get_center('z')
    x1 = sim.setup.mesher.get_min('x')
    x2 = sim.setup.mesher.get_max('x')
    points = [[x1, y, z], [x2, y, z]]

    sim.add_postprocess(get_postprocess_path("pipe_postprocess.py"))
    sim.run_postprocess('plot_pressure',
                        worker_name='local',
                        args={'points': points,
                              'folder': sim.load_path('local')})

app.add_main(my_fun)

if __name__ == '__main__':
    args = app.parse_inputs()
    app.run(args)

Severe Slug¶

import transat.appbuilder as appbuilder

from transat.simulation import Simulation

from transat.setup.cad import CAD
from transat.setup.cad import Pipe
from transat.setup.cad import Network

from transat.server.server import Server
import threading
from transat.server.client import Client

import math
import time
import os

from progressbar import Bar, ETA, Percentage, ProgressBar, ReverseBar, RotatingMarker

from transat.config import ascomp_setup as setup

global_config = setup.install()

app = appbuilder.App("Severe Slug")
app.add_input("L1", unit="m")
app.add_input("L2", unit="m")
app.add_input("L3", unit="m")
app.add_input("OilVel", unit="m/s")
app.add_input("WaterVel", unit="m/s")
app.add_input("GasVel", unit="m/s")
app.add_input("Radius", unit="m")

def get_postprocess_path(name):
    path = os.path.realpath(__file__)
    path = os.path.dirname(path)
    return os.path.join(path, name)

def create_pipe(L1, L2, L3, pipe_radius=0.25):
    cad = CAD(freecad_path=global_config.env['path']['freecad'])

    pipe = Pipe(name="pipe.stl", radius=pipe_radius)
    pipe.add_point([0.0, 0.0, 0])
    pipe.add_point([L1, 0.0, 0], 0.1)
    pipe.add_point([L1, L2, 0], 0.1)
    pipe.add_point([L1+L3, L2, 0])

    network = Network(cad, [pipe])

    return network, pipe

def launch_server():
    se = Server()
    server = threading.Thread(target=se.run)
    server.daemon = True
    server.start()
    return se.get_address()



def watch(url):

    widgets = ['Simulation: ', Percentage(), ' ', Bar(marker=RotatingMarker()),
           ' ', ETA(), ' ']
    pbar = ProgressBar(widgets=widgets, maxval=100).start()

    start = time.time()
    cl = Client(url)
    iteration = None
    while cl.has_jobs():
        elapsed = (time.time() - start)
        remaining = cl.get_remaining_time()*60.0
        percentage = elapsed/(elapsed+remaining)*100
        pbar.update(percentage)
        _iteration = cl.get_current_timestep()
        if iteration != _iteration:
            iteration = _iteration
        time.sleep(1)
    pbar.finish()
    print cl.get_current_timestep()


def my_fun(inputs):
    url = launch_server()

    radius = float(inputs['Radius'])
    L1 = float(inputs['L1'])
    L2 = float(inputs['L2'])
    L3 = float(inputs['L3'])

    sim = Simulation("Severe_Slug")
    # simulations is saved in {local_wd}/Pipe/test_pipe
    sim.set_folder('2D_3', ['local'])

    # load templates. It will first look in the relative directory
    # and then in the installation directory
    sim.load_template('templates/severe_slug')

    network, pipe = create_pipe(L1, L2, L3, radius)
    network.create(path=sim.load_path('local'))

    sim.setup.mesher.set_blocks(network.get_blocks())
    sim.setup.mesher.set_dx(radius / 5.0)
    sim.setup.ist.add(network.get_pipes_name())


    gas = sim.setup.input.get_phases().get_phase("Gas")
    gas.set_density(0.71)
    oil = sim.setup.input.get_phases().get_phase("Oil")
    water = sim.setup.input.get_phases().get_phase("Water")

    sim.get_bc('inflow').set_phase_velocities(water, velocity=[float(inputs['WaterVel']), 0, 0], volume_fraction=0.2)
    sim.get_bc('inflow').set_phase_velocities(oil, velocity=[ float(inputs['OilVel']) , 0, 0], volume_fraction=0.5)
    sim.get_bc('inflow').set_phase_velocities(gas, velocity=[ float(inputs['GasVel']), 0, 0], volume_fraction=0.3)

    # is not going to chage the initial conditions. Convergence may be difficult
    sim.get_bc('outflow').set_pressure(0.0)

    tree = sim.setup.bcs.create_surfaces_tree()
    tree.get_all_surfaces().set_bc_name('wall')

    tree.get_surface_with_point(pipe.get_inlet_point()).set_bc_name('inflow')
    tree.get_surface_with_point(pipe.get_outlet_point()).set_bc_name('outflow')

    offset = 1e-7
    sim.setup.mesher.set_2D('x', 'y', offset)
    tree.get_surface_with_normal('z').set_bc_name('symmetry')

    sim.prepare_simulation_files()

    sim.run_init(worker='local', nprocs=3)
    #sim.run(worker='local', nprocs=3, url=url)
    sim.run_until(time=3, worker='local', nprocs=3, url=url)

    watch(url)

    A_2d = 2*radius*1
    A_3d = math.pi*math.pow(radius, 2)

    print sim.postprocess.get_massflow_rate_2D(sim.load_path('local'), A_2d, A_3d, [gas, water, oil], pipe.get_outlet_point(), [1,0,0])
    print sim.postprocess.get_pressure_drop(sim.load_path('local'), [0.03, 0, 0], pipe.get_outlet_point())

    print sim.load_path('local')

app.add_main(my_fun)

if __name__ == "__main__":
    inputs = app.parse_inputs()
    app.run(inputs)

Pipe Network¶

from transat.simulation import Simulation
import transat.appbuilder as appbuilder
from transat.setup.cad import Pipe
from transat.setup.cad import Network
from transat.setup.cad import CAD
from transat.config import ascomp_setup as setup

global_config = setup.install()
app = appbuilder.App('Network')


def create_network(pipe_radius = 0.25, elbow_radius = 1):
    cad = CAD(freecad_path=global_config.env['path']['freecad'])

    pipe = Pipe(name="pipe.stl", radius=pipe_radius)
    pipe.add_point([-2.5, 0.0, 0])
    pipe.add_point([2.5, 0.0, 0], elbow_radius)
    pipe.add_point([2.5, -2.5, 0], elbow_radius)
    pipe.add_point([5.0, -2.5, 0], elbow_radius)
    pipe.add_point([5.0, 0.0, 0], elbow_radius)
    pipe.add_point([7.5, 0.0, 0])

    elbow_radius = 1
    pipe2 = Pipe(name="pipe2.stl", radius=pipe_radius)
    pipe2.add_point([0, 0.0, -2.5])
    pipe2.add_point([0, 2.5, -2.5], elbow_radius)
    pipe2.add_point([0, 2.5, 0], elbow_radius)
    pipe2.add_point([0, 0.0, 0])

    network = Network(cad, [pipe2, pipe])
    return network, pipe, pipe2

def my_fun(args):

    sim = Simulation('Network')
    sim.set_folder('network2', ['local', 'calanda'])
    sim.load_template('templates/network')

    pipe_radius = 0.25
    elbow_radius = 0.5
    network, pipe, pipe2 = create_network(pipe_radius, elbow_radius)
    network.create(sim.load_path('local'))

    sim.setup.mesher.set_blocks(network.get_blocks())
    sim.setup.mesher.split_blocks()
    sim.setup.ist.add(network.get_pipes_name())

    sim.setup.mesher.set_dx(pipe_radius / 5.0)

    sim.setup.bcs.get_bc('inflow_gas').set_velocities([0.0, 20.0, 0])
    sim.setup.bcs.get_bc('inflow_liq').set_velocities([1.0, 0, 0])

    tree = sim.setup.bcs.create_surfaces_tree()
    tree.get_all_surfaces().set_bc_name('wall')

    tree.get_surface_with_point(pipe.get_inlet_point()).set_bc_name('inflow_liq')
    tree.get_surface_with_point(pipe.get_outlet_point()).set_bc_name('outflow')
    tree.get_surface_with_point(pipe2.get_inlet_point()).set_bc_name('inflow_gas')

    sim.prepare_simulation_files()

    n = sim.setup.mesher.get_number_of_blocks()

    sim.run_init(worker='calanda', nprocs=8)
    sim.run(worker='calanda', nprocs=8)

    return True


app.add_main(my_fun)

if __name__ == "__main__":
    inputs = app.parse_inputs()
    app.run(inputs)

Simulation class¶

The high-level access to TransAT is typically through the Simulation class. Instances of this class have access to top-level functions to define, configure, start, stop and postprocess TransAT simulations.

To create a new simulation object:

>>> from transat.simulation import Simulation
>>> sim = Simulation()

To load an existing setup (typically created using TransAT GUI):

>>> sim.load_template('/path/to/project/directory')

To start a simulation on the local workstation:

>>> sim.run()

To start a simulation on a cluster using 64 processors:

>>> sim.run(worker='cluster', nprocs=64)

where 'cluster' has been configured in the Configuration class. The list of available functions is documented here.

Boundary conditions¶

The boundary conditions in TransAT are described through the Boundary Conditions class and can be manipulated programmatically.

Assuming we have created a setup using the GUI and stored it in a directory name my_template:

>>> from transat.simulation import Simulation
>>> sim = Simulation()
>>> sim.load_template('my_template')

We can get the list of all boundary conditions in that setup:

>>> sim.get_bc_names()
['North', 'South', 'East', 'West']

Any of these boundary conditions can be modified:

>>> sim.get_bc('East').set_velocity(3.5)

The list of available functions is documented here.

Application builder¶

To reduce boiler-plate code, an Application builder wrapper module is provided. It allows for fast prototyping of Python applications that expose only a very simple API to the user.

Packages contents¶

transat package