# -*- coding: utf-8 -*-
"""Base Class for an OGS5 run."""
import glob
import os
import shutil
import sys
import time
import warnings
from copy import deepcopy as dcp
import pexpect
from pexpect.popen_spawn import PopenSpawn
from ogs5py.fileclasses import (
ASC,
BC,
CCT,
DDC,
FCT,
GEM,
GLI,
IC,
KRC,
MCP,
MFP,
MMP,
MPD,
MSH,
MSP,
NUM,
OUT,
PCS,
PCT,
PQC,
REI,
RFD,
RFR,
ST,
TIM,
GEMinit,
GLIext,
PQCdat,
)
from ogs5py.fileclasses.base import BOT_COM, CWD, TOP_COM, MultiFile
from ogs5py.tools.download import OGS5PY_CONFIG
from ogs5py.tools.script import gen_script
from ogs5py.tools.tools import Output, search_task_id
from ogs5py.tools.types import MULTI_FILES, OGS_EXT
# pexpect.spawn just runs on unix-like systems
if sys.platform == "win32":
OGS_NAME = "ogs.exe"
CmdRun = PopenSpawn
else:
OGS_NAME = "ogs"
CmdRun = pexpect.spawn
[docs]class OGS:
"""Class for an OGS5 model.
In this class everything for an OGS5 model can be specified.
Parameters
----------
task_root : :class:`str`, optional
Path to the destiny model folder.
Default: cwd+"ogs5model"
task_id : :class:`str`, optional
Name for the ogs task.
Default: "model"
output_dir : :class:`str` or :class:`None`, optional
Path to the output directory.
Default: :class:`None`
Notes
-----
The following Classes are present as attributes
bc : Boundary Condition
Information of the Boundary Conditions for the model.
cct : Communication Table
Information of the Communication Table for the model.
fct : Function
Information of the Function definitions for the model.
gem : geochemical thermodynamic modeling coupling
Information of the geochemical thermodynamic modeling
coupling for the model.
gli : Geometry
Information of the Geometry for the model.
ic : Initial Condition
Information of the Initial Conditions for the model.
krc : Kinetric Reaction
Information of the Kinetric Reaction for the model.
mcp : reactive components for modelling chemical processes
Information of the reactive components for
modelling chemical processes for the model.
mfp : Fluid Properties
Information of the Fluid Properties for the model.
mmp : Medium Properties
Information of the Medium Properties for the model.
msh : Mesh
Information of the Mesh for the model.
msp : Solid Properties
Information of the Solid Properties for the model.
num : Settings for the numerical solver
Information of the numerical solver for the model.
out : Output Settings
Information of the Output Settings for the model.
pcs : Process settings
Information of the Process settings for the model.
pct : Particle Definition for Random walk
Information of the Particles defined for Randomwalk setting.
pqc : Phreqqc coupling (not supported yet)
Information of the Boundary Conditions for the model.
pqcdat : Phreqqc coupling (the phreeqc.dat file)
phreeqc.dat file for the model.
(just a line-wise file with no comfort)
rei : Reaction Interface
Information of the Reaction Interface for the model.
rfd : definition of time-curves for variing BCs or STs
Information of the time curves for the model.
st : Source Term
Information of the Source Term for the model.
tim : Time settings
Information of the Time settings for the model.
Additional
mpd : Distributed Properties (list of files)
Information of the Distributed Properties for the model.
gli_ext : list for external Geometry definition
External definition of surfaces (TIN) or polylines (POINT_VECTOR)
rfr : list of restart files
RESTART files as defined in the INITIAL_CONDITION
gem_init : list of GEMS3K input files (lst file)
given as GEMinit classes
asc : list of ogs ASC files
This file type comes either from .tim .pcs or .gem
copy_files : list of path-strings
Files that should be copied to the destiny folder.
"""
def __init__(self, task_root=None, task_id="model", output_dir=None):
if task_root is None:
task_root = os.path.join(CWD, "ogs5model")
self._task_root = os.path.normpath(task_root)
self._task_id = task_id
self._output_dir = None
self.output_dir = output_dir
self.exitstatus = None
self.bc = BC(task_root=task_root, task_id=task_id)
self.cct = CCT(task_root=task_root, task_id=task_id)
self.ddc = DDC(task_root=task_root, task_id=task_id)
self.fct = FCT(task_root=task_root, task_id=task_id)
self.gem = GEM(task_root=task_root, task_id=task_id)
self.gli = GLI(task_root=task_root, task_id=task_id)
self.ic = IC(task_root=task_root, task_id=task_id)
self.krc = KRC(task_root=task_root, task_id=task_id)
self.mcp = MCP(task_root=task_root, task_id=task_id)
self.mfp = MFP(task_root=task_root, task_id=task_id)
self.mmp = MMP(task_root=task_root, task_id=task_id)
self.msh = MSH(task_root=task_root, task_id=task_id)
self.msp = MSP(task_root=task_root, task_id=task_id)
self.num = NUM(task_root=task_root, task_id=task_id)
self.out = OUT(task_root=task_root, task_id=task_id)
self.pcs = PCS(task_root=task_root, task_id=task_id)
self.pct = PCT(task_root=task_root, task_id=task_id)
self.pqc = PQC(task_root=task_root, task_id=task_id)
self.pqcdat = PQCdat(task_root=task_root, task_id=task_id)
self.rei = REI(task_root=task_root, task_id=task_id)
self.rfd = RFD(task_root=task_root, task_id=task_id)
self.st = ST(task_root=task_root, task_id=task_id)
self.tim = TIM(task_root=task_root, task_id=task_id)
# create a list for mpd files
self.mpd = MultiFile(MPD, task_root=task_root, task_id=task_id)
# create a list for external Geometry definition (TIN and POINT_VECTOR)
self.gli_ext = MultiFile(GLIext, task_root=task_root, task_id=task_id)
# create a list for RESTART files in the INITIAL_CONDITION
self.rfr = MultiFile(RFR, task_root=task_root, task_id=task_id)
# create a list for GEMS3K input files (lst file)
self.gem_init = MultiFile(
GEMinit, task_root=task_root, task_id=task_id
)
# create a list for ASC files
self.asc = MultiFile(ASC, task_root=task_root, task_id=task_id)
# create a list of arbitrary files to be copied (names will be same)
self.copy_files = []
# store the Top Comment
self._top_com = TOP_COM
# store the Bottom Comment
self._bot_com = BOT_COM
@property
def top_com(self):
"""Get and set the top comment for the ogs files."""
return self._top_com
@top_com.setter
def top_com(self, value):
self._top_com = value
for ext in OGS_EXT:
# workaround to get access to class-members by name
getattr(self, ext[1:]).top_com = value
@property
def bot_com(self):
"""Get and set the bottom comment for the ogs files."""
return self._bot_com
@bot_com.setter
def bot_com(self, value):
self._bot_com = value
for ext in OGS_EXT:
# workaround to get access to class-members by name
getattr(self, ext[1:]).bot_com = value
@property
def task_root(self):
"""Get and set the task_root path of the ogs model."""
return self._task_root
@task_root.setter
def task_root(self, value):
self._task_root = value
for ext in OGS_EXT:
# workaround to get access to class-members by name
getattr(self, ext[1:]).task_root = value
for ext in MULTI_FILES:
getattr(self, ext).standard["task_root"] = value
for i in range(len(getattr(self, ext))):
getattr(self, ext)[i].task_root = value
self.pqcdat.task_root = value
@property
def task_id(self):
""":class:`str`: task_id (name) of the ogs model."""
return self._task_id
@task_id.setter
def task_id(self, value):
# workaround for asc
for i in range(len(self.asc)):
self.asc[i].name = value + self.asc[i].name[len(self._task_id) :]
self._task_id = value
for ext in OGS_EXT:
getattr(self, ext[1:]).task_id = value
for ext in MULTI_FILES:
getattr(self, ext).standard["task_id"] = value
@property
def output_dir(self):
""":class:`str`: output directory path of the ogs model."""
return self._output_dir
@output_dir.setter
def output_dir(self, value):
if value is None:
self._output_dir = None
else:
self._output_dir = os.path.normpath(value)
if not os.path.isabs(self._output_dir):
# if not, put the outputfolder in the task_root
self._output_dir = os.path.join(
os.path.abspath(self.task_root), self._output_dir
)
@property
def has_output_dir(self):
""":class:`bool`: State if the model has a output directory."""
return self.output_dir is not None
[docs] def add_copy_file(self, path):
"""
Method to add an arbitrary file that should be copied.
The base-name of the file will be keept and it will be copied to
the task-root when the "write" routine is called.
"""
if os.path.isfile(path):
self.copy_files.append(os.path.abspath(path))
else:
print("OGS.add_copy_file: given file is not valid: " + str(path))
[docs] def del_copy_file(self, index=None):
"""
Method to delete a copy-file.
Parameters
----------
index : int or None, optional
The index of the copy-file that should be deleted. If None, all
copy-files are deleted. Default: None
"""
if index is None:
self.copy_files = []
elif -len(self.copy_files) <= index < len(self.copy_files):
del self.copy_files[index]
else:
print("OGS.del_copy_file: given index is not valid.")
[docs] def add_mpd(self, mpd_file):
"""
Method to add an ogs MEDIUM_PROPERTIES_DISTRIBUTED file to the model.
This is used for disributed information in the MMP file.
See ogs5py.MPD for further information
"""
if isinstance(mpd_file, MPD):
mpd_file.task_root = self.task_root
self.mpd.append(mpd_file)
[docs] def del_mpd(self, index=None):
"""
Method to delete MEDIUM_PROPERTIES_DISTRIBUTED file.
Parameters
----------
index : int or None, optional
The index of the mpd-file that should be deleted. If None, all
mpd-files are deleted. Default: None
"""
if index is None:
self.mpd.reset_all()
elif -len(self.mpd) <= index < len(self.mpd):
del self.mpd[index]
else:
print("OGS.del_mpd: given index is not valid.")
[docs] def add_gli_ext(self, gli_ext_file):
"""
Method to add an external Geometry definition file to the model.
This is used for TIN definition in SURFACE or POINT_VECTOR definition
in POLYLINE in the GLI file.
See ogs5py.GLI for further information
"""
if isinstance(gli_ext_file, GLIext):
gli_ext_file.task_root = self.task_root
self.gli_ext.append(gli_ext_file)
[docs] def del_gli_ext(self, index=None):
"""
Method to delete external Geometry file.
Parameters
----------
index : int or None, optional
The index of the external gli file that should be deleted.
If None, all external gli files are deleted. Default: None
"""
if index is None:
self.gli_ext.reset_all()
elif -len(self.gli_ext) <= index < len(self.gli_ext):
del self.gli_ext[index]
else:
print("OGS.del_gli_ext: given index is not valid.")
[docs] def add_rfr(self, rfr_file):
"""
Method to add an ogs RESTART file to the model.
This is used for disributed information in the IC file.
See ogs5py.IC for further information
"""
if isinstance(rfr_file, RFR):
rfr_file.task_root = self.task_root
self.rfr.append(rfr_file)
[docs] def del_rfr(self, index=None):
"""
Method to delete RESTART file.
Parameters
----------
index : int or None, optional
The index of the RESTART file that should be deleted.
If None, all RESTART files are deleted. Default: None
"""
if index is None:
self.rfr.reset_all()
elif -len(self.rfr) <= index < len(self.rfr):
del self.rfr[index]
else:
print("OGS.del_rfr: given index is not valid.")
[docs] def add_gem_init(self, gem_init_file):
"""
Method to add a GEMS3K input file.
This is usually generated by GEM-SELEKTOR.
See ogs5py.GEM and ogs5py.GEMinit for further information
"""
if isinstance(gem_init_file, GEMinit):
gem_init_file.task_root = self.task_root
self.gem_init.append(gem_init_file)
[docs] def del_gem_init(self, index=None):
"""
Method to delete GEMS3K input file.
Parameters
----------
index : int or None, optional
The index of the GEMS3K file that should be deleted.
If None, all GEMS3K files are deleted. Default: None
"""
if index is None:
self.gem_init.reset_all()
elif -len(self.gem_init) <= index < len(self.gem_init):
del self.gem_init[index]
else:
print("OGS.del_rfr: given index is not valid.")
[docs] def add_asc(self, asc_file):
"""
Method to add a ASC file.
See ogs5py.ASC for further information
"""
if isinstance(asc_file, ASC):
asc_file.task_root = self.task_root
self.asc.append(asc_file)
[docs] def del_asc(self, index=None):
"""
Method to delete a ASC file.
Parameters
----------
index : int or None, optional
The index of the ASC file that should be deleted.
If None, all ASC files are deleted. Default: None
"""
if index is None:
self.asc.reset_all()
elif -len(self.asc) <= index < len(self.asc):
del self.asc[index]
else:
print("OGS.del_rfr: given index is not valid.")
[docs] def reset(self):
"""Delete every content."""
for ext in OGS_EXT:
# workaround to get access to class-members by name
getattr(self, ext[1:]).reset()
for ext in MULTI_FILES:
getattr(self, ext).reset_all()
self.pqcdat.reset()
self.copy_files = []
# reset to initial attributes
self.task_root = self._task_root
self.task_id = self._task_id
self.top_com = self._top_com
self.bot_com = self._bot_com
[docs] def gen_script(
self,
script_dir=os.path.join(os.getcwd(), "ogs_script"),
script_name="model.py",
ogs_cls_name="model",
task_root=None,
task_id=None,
output_dir=None,
separate_files=None,
):
"""
Generate a python script for the given model.
Parameters
----------
script_dir : str
target directory for the script
script_name : str
name for the script file (including .py ending)
ogs_cls_name : str
name of the model in the script
task_root : str
used task_root in the script
task_id : str
used task_id in the script
output_dir : str
used output_dir in the script
separate_files : list of str or None
list of files, that should be written to separate files and
then loaded from the script
Notes
-----
This will only create BlockFiles from the script. GLI and MSH files
as well as every other file are stored separately.
"""
gen_script(
self,
script_dir,
script_name,
ogs_cls_name,
task_root,
task_id,
output_dir,
separate_files,
)
[docs] def load_model(
self,
task_root,
task_id=None,
use_task_root=False,
use_task_id=False,
skip_files=None,
skip_ext=None,
encoding=None,
verbose=False,
search_ext=None, # (".pcs"),
):
"""
Load an existing OGS5 model.
Parameters
----------
task_root : str
Path to the destiny folder.
task_id : str or None, optional
Task ID of the model to load. If None is given, it will be
determind by the found files. If multiple possible task_ids were
found, the first one in alphabetic order will be used.
Default: None
use_task_root : Bool, optional
State if the given task_root should be used for this model.
Default: False
use_task_id : Bool, optional
State if the given task_id should be used for this model.
Default: False
skip_files : list or None, optional
List of file-names, that should not be read. Default: None
skip_ext : list or None, optional
List of file-extensions, that should not be read. Default: None
encoding : str or None, optional
encoding of the given files. If ``None`` is given, the system
standard is used. Default: ``None``
verbose : bool, optional
Print information of the reading process. Default: False
search_ext : str
OGS extension that should be searched for. Default: ".pcs"
Notes
-----
This method will search for all known OGS5 file-extensions in the
given path (task_root).
Additional files from:
- GLI (POINT_VECTOR + TIN)
- MMP (distributed media properties)
- IC (RESTART)
- GEM (GEM3SK init file)
will be read automatically.
If you get an ``UnicodeDecodeError`` try loading with:
``encoding="ISO-8859-15"``
"""
self.reset()
if skip_files is None:
skip_files = []
if skip_ext is None:
skip_ext = []
# search for possible task_ids in the directory
found_ids = search_task_id(task_root, search_ext)
if not found_ids:
raise ValueError(
"ogs5py.OGS.laod_model - nothing was found at: " + task_root
)
if verbose:
print("ogs5py.OGS.laod_model - found ids: " + str(found_ids))
# take the first found task_id
if task_id is None:
# take the first found task_id
task_id = found_ids[0]
# check if the given task_id is found
elif task_id not in found_ids:
raise ValueError(
"ogs5py.OGS.load_model - didn't find given task_id ("
+ task_id
+ ") at: "
+ task_root
)
if verbose:
print("ogs5py.OGS.laod_model - use id: " + task_id)
# overwrite the task_root
if use_task_root:
self.task_root = task_root
# overwrite the task_id
if use_task_id:
self.task_id = task_id
# iterate over all ogs file-extensions
for ext in OGS_EXT:
if verbose:
print(ext, end=" ")
# skip certain file extensions if wanted
if ext in skip_ext or ext[1:] in skip_ext:
continue
# search for files with given extension
files = glob.glob(os.path.join(task_root, task_id + ext))
# if nothing was found, skip
if not files:
continue
# take the first found file if there are multiple
fil = files[0]
# skip file if wanted
if os.path.basename(fil) in skip_files or fil in skip_files:
continue
# workaround to get access to class-members by name
getattr(self, ext[1:]).read_file(
fil, encoding=encoding, verbose=verbose
)
# append GEOMETRY defnitions
if ext == ".gli":
for ply in self.gli.POLYLINES:
# POINT_VECTOR definition of a POLYLINE
ext_name = ply["POINT_VECTOR"]
if ext_name is not None:
raw_name = os.path.basename(ext_name)
f_name, f_ext = os.path.splitext(raw_name)
ext_file = GLIext(
typ="POINT_VECTOR",
name=f_name,
file_ext=f_ext,
task_root=self.task_root,
)
path = os.path.join(task_root, ext_name)
ext_file.read_file(path, encoding=encoding)
self.gli_ext.append(dcp(ext_file))
for srf in self.gli.SURFACES:
# Triangulation definition of a SURFACE
ext_name = srf["TIN"]
if ext_name is not None:
raw_name = os.path.basename(ext_name)
f_name, f_ext = os.path.splitext(raw_name)
ext_file = GLIext(
typ="TIN",
name=f_name,
file_ext=f_ext,
task_root=self.task_root,
)
path = os.path.join(task_root, ext_name)
ext_file.read_file(path, encoding=encoding)
self.gli_ext.append(dcp(ext_file))
# append MEDIUM_PROPERTIES_DISTRIBUTED defnitions
if ext == ".mmp":
for i in range(len(self.mmp.mainkw)):
# external PERMEABILITY_DISTRIBUTION
if "PERMEABILITY_DISTRIBUTION" in self.mmp.subkw[i]:
index = self.mmp.subkw[i].index(
"PERMEABILITY_DISTRIBUTION"
)
ext_name = self.mmp.cont[i][index][0][0]
raw_name = os.path.basename(ext_name)
f_name, f_ext = os.path.splitext(raw_name)
ext_file = MPD(
name=f_name,
file_ext=f_ext,
task_root=self.task_root,
)
path = os.path.join(task_root, ext_name)
ext_file.read_file(path, encoding=encoding)
self.mpd.append(dcp(ext_file))
# external POROSITY_DISTRIBUTION
if "POROSITY_DISTRIBUTION" in self.mmp.subkw[i]:
index = self.mmp.subkw[i].index(
"POROSITY_DISTRIBUTION"
)
ext_name = self.mmp.cont[i][index][0][0]
raw_name = os.path.basename(ext_name)
f_name, f_ext = os.path.splitext(raw_name)
ext_file = MPD(
name=f_name,
file_ext=f_ext,
task_root=self.task_root,
)
path = os.path.join(task_root, ext_name)
ext_file.read_file(path, encoding=encoding)
self.mpd.append(dcp(ext_file))
# external GEOMETRY_AREA
if "GEOMETRY_AREA" in self.mmp.subkw[i]:
index = self.mmp.subkw[i].index("GEOMETRY_AREA")
if self.mmp.cont[i][index][0][0] == "FILE":
ext_name = self.mmp.cont[i][index][0][1]
raw_name = os.path.basename(ext_name)
f_name, f_ext = os.path.splitext(raw_name)
ext_file = MPD(
name=f_name,
file_ext=f_ext,
task_root=self.task_root,
)
path = os.path.join(task_root, ext_name)
ext_file.read_file(path, encoding=encoding)
self.mpd.append(dcp(ext_file))
# append GEMS3K init file
if ext == ".gem":
for i in range(len(self.gem.mainkw)):
if "GEM_INIT_FILE" in self.gem.subkw[i]:
index = self.gem.subkw[i].index("GEM_INIT_FILE")
ext_name = self.gem.cont[i][index][0][0]
ext_file = GEMinit(
lst_name=ext_name, task_root=self.task_root
)
path = os.path.join(task_root, ext_name)
ext_file.read_file(path, encoding=encoding)
self.gem_init.append(dcp(ext_file))
# append RESART defnitions
if ext == ".ic":
for i in range(len(self.ic.mainkw)):
if "DIS_TYPE" in self.ic.subkw[i]:
index = self.ic.subkw[i].index("DIS_TYPE")
if self.ic.cont[i][index][0][0] != "RESTART":
continue
ext_name = self.ic.cont[i][index][0][1]
raw_name = os.path.basename(ext_name)
f_name, f_ext = os.path.splitext(raw_name)
ext_file = RFR(
name=f_name,
file_ext=f_ext,
task_root=self.task_root,
)
path = os.path.join(task_root, ext_name)
ext_file.read_file(path, encoding=encoding)
self.rfr.append(dcp(ext_file))
# read phreeqc.dat
if ext == ".pqc": # phreeqc.dat or Phreeqc.dat
pqcfiles = glob.glob(os.path.join(task_root, "*hreeqc.dat"))
self.pqcdat.read_file(
path=pqcfiles[0], encoding=encoding, verbose=verbose
)
# load ASC files
files = glob.glob(os.path.join(task_root, task_id + "*.asc"))
for fil in files:
raw_name = os.path.basename(fil)
f_name, f_ext = os.path.splitext(raw_name)
ext_file = ASC(
name=self.task_id + f_name[len(task_id) :],
task_root=self.task_root,
)
path = os.path.join(task_root, fil)
ext_file.read_file(path, encoding=encoding)
self.asc.append(dcp(ext_file))
return True
[docs] def readvtk(self, pcs="ALL", output_dir=None):
r"""
Reader for vtk outputfiles of this OGS5 model.
Parameters
----------
pcs : string or None, optional
specify the PCS type that should be collected
Possible values are:
- None/"" (no PCS_TYPE specified in \*.out)
- "NO_PCS"
- "GROUNDWATER_FLOW"
- "LIQUID_FLOW"
- "RICHARDS_FLOW"
- "AIR_FLOW"
- "MULTI_PHASE_FLOW"
- "PS_GLOBAL"
- "HEAT_TRANSPORT"
- "DEFORMATION"
- "MASS_TRANSPORT"
- "OVERLAND_FLOW"
- "FLUID_MOMENTUM"
- "RANDOM_WALK"
You can get a list with all known PCS-types by setting PCS="ALL"
Default : None
output_dir : :any:'None' or :class:'str', optional
Sometimes OGS5 doesn't put the output in the right directory.
You can specify a separate output directory here in this case.
Default: :any:'None'
Returns
-------
result : dict
keys are the point names and the items are the data from the
corresponding files
if pcs="ALL", the output is a dictionary with the PCS-types as keys
"""
from ogs5py.reader import readvtk as read
if output_dir is not None:
root = output_dir
elif self.has_output_dir:
root = self.output_dir
else:
root = self.task_root
return read(task_root=root, task_id=self.task_id, pcs=pcs)
[docs] def readpvd(self, pcs="ALL", output_dir=None):
r"""
Read the paraview pvd files of this OGS5 model.
All concerned files are converted to a dictionary containing their data
Parameters
----------
pcs : string or None, optional
specify the PCS type that should be collected
Possible values are:
- None/"" (no PCS_TYPE specified in \*.out)
- "NO_PCS"
- "GROUNDWATER_FLOW"
- "LIQUID_FLOW"
- "RICHARDS_FLOW"
- "AIR_FLOW"
- "MULTI_PHASE_FLOW"
- "PS_GLOBAL"
- "HEAT_TRANSPORT"
- "DEFORMATION"
- "MASS_TRANSPORT"
- "OVERLAND_FLOW"
- "FLUID_MOMENTUM"
- "RANDOM_WALK"
You can get a list with all known PCS-types by setting PCS="ALL"
Default : "ALL"
output_dir : :any:'None' or :class:'str', optional
Sometimes OGS5 doesn't put the output in the right directory.
You can specify a separate output directory here in this case.
Default: :any:'None'
Returns
-------
result : dict
keys are the point names and the items are the data from the
corresponding files
if pcs="ALL", the output is a dictionary with the PCS-types as keys
"""
from ogs5py.reader import readpvd as read
if output_dir is not None:
root = output_dir
elif self.has_output_dir:
root = self.output_dir
else:
root = self.task_root
return read(task_root=root, task_id=self.task_id, pcs=pcs)
[docs] def readtec_point(self, pcs="ALL", output_dir=None):
r"""
Collect TECPLOT point output from this OGS5 model.
Parameters
----------
pcs : string or None, optional
specify the PCS type that should be collected
Possible values are:
- None/"" (no PCS_TYPE specified in \*.out)
- "NO_PCS"
- "GROUNDWATER_FLOW"
- "LIQUID_FLOW"
- "RICHARDS_FLOW"
- "AIR_FLOW"
- "MULTI_PHASE_FLOW"
- "PS_GLOBAL"
- "HEAT_TRANSPORT"
- "DEFORMATION"
- "MASS_TRANSPORT"
- "OVERLAND_FLOW"
- "FLUID_MOMENTUM"
- "RANDOM_WALK"
You can get a list with all known PCS-types by setting PCS="ALL"
Default : "ALL"
output_dir : :any:'None' or :class:'str', optional
Sometimes OGS5 doesn't put the output in the right directory.
You can specify a separate output directory here in this case.
Default: :any:'None'
Returns
-------
result : dict
Keys are the point names and the items are the data from the
corresponding files. If pcs="ALL",
the output is a dictionary with the PCS-types as keys.
"""
from ogs5py.reader import readtec_point as read
if output_dir is not None:
root = output_dir
elif self.has_output_dir:
root = self.output_dir
else:
root = self.task_root
return read(task_root=root, task_id=self.task_id, pcs=pcs)
[docs] def readtec_polyline(self, pcs="ALL", trim=True, output_dir=None):
r"""
Collect TECPLOT polyline output from this OGS5 model.
Parameters
----------
pcs : string or None, optional
specify the PCS type that should be collected
Possible values are:
- None/"" (no PCS_TYPE specified in \*.out)
- "NO_PCS"
- "GROUNDWATER_FLOW"
- "LIQUID_FLOW"
- "RICHARDS_FLOW"
- "AIR_FLOW"
- "MULTI_PHASE_FLOW"
- "PS_GLOBAL"
- "HEAT_TRANSPORT"
- "DEFORMATION"
- "MASS_TRANSPORT"
- "OVERLAND_FLOW"
- "FLUID_MOMENTUM"
- "RANDOM_WALK"
You can get a list with all known PCS-types by setting pcs="ALL"
Default : "ALL"
output_dir : :any:'None' or :class:'str', optional
Sometimes OGS5 doesn't put the output in the right directory.
You can specify a separate output directory here in this case.
Default: :any:'None'
trim : Bool, optional
if the ply_ids are not continuous, there will be "None" values in
the output list. If trim is "True" these values will be eliminated.
If there is just one output for a polyline, the list will
be eliminated and the output will be the single dict.
Default : True
Returns
-------
result : dict
keys are the Polyline names and the items are lists sorted by the
ply_id (it is assumed, that the ply_ids are continuous, if not, the
corresponding list entries are "None")
if pcs="ALL", the output is a dictionary with the PCS-types as keys
"""
from ogs5py.reader import readtec_polyline as read
if output_dir is not None:
root = output_dir
elif self.has_output_dir:
root = self.output_dir
else:
root = self.task_root
return read(task_root=root, task_id=self.task_id, pcs=pcs, trim=trim)
[docs] def output_files(self, pcs=None, typ="VTK", element=None, output_dir=None):
r"""
Get a list of output file paths.
Parameters
----------
pcs : string or None, optional
specify the PCS type that should be collected
Possible values are:
- None/"" (no PCS_TYPE specified in \*.out)
- "NO_PCS"
- "GROUNDWATER_FLOW"
- "LIQUID_FLOW"
- "RICHARDS_FLOW"
- "AIR_FLOW"
- "MULTI_PHASE_FLOW"
- "PS_GLOBAL"
- "HEAT_TRANSPORT"
- "DEFORMATION"
- "MASS_TRANSPORT"
- "OVERLAND_FLOW"
- "FLUID_MOMENTUM"
- "RANDOM_WALK"
Default : None
typ : string, optional
Type of the output ("VTK", "PVD", "TEC_POINT" or "TEC_POLYLINE").
Default : "VTK"
element : string or None, optional
For tecplot output you can specify the name of the output element.
(Point-name of Line-name from GLI file)
Default: None
"""
from ogs5py.tools.output import get_output_files as read
if output_dir is not None:
root = output_dir
elif self.has_output_dir:
root = self.output_dir
else:
root = self.task_root
return read(root, self.task_id, pcs, typ, element)
[docs] def run_model(
self,
ogs_exe=None,
ogs_name=OGS_NAME,
print_log=True,
save_log=True,
log_path=None,
log_name=None,
timeout=None,
):
"""
Run the defined OGS5 model.
Parameters
----------
ogs_exe : str or None, optional
path to the ogs executable. If ``None`` is given, the default sys
path will be searched with ``which``.
Can be a folder containing the exe with basename: ogs_name.
It will first look in the :any:`OGS5PY_CONFIG` folder.
Default: None
ogs_name : str or None, optional
Name of to the ogs executable to search for.
Just used if ,ogs_exe is ``None``. Default: ``"ogs"``
print_log : bool, optional
state if the ogs output should be displayed in the terminal.
Default: True
save_log : bool, optional
state if the ogs output should be saved to a file.
Default: True
log_path : str or None, optional
Path, where the log file should be saved. Default: None
(the defined output directory or the task_root directory)
log_name : str or None, optional
Name of the log file. Default: None
(task_id+time+"_log.txt")
timeout : int or None, optional
Time to wait for OGS5 to finish in seconds. Default: None
Returns
-------
success : bool
State if OGS5 terminated 'normally'. (Allways true on Windows.)
"""
# look for the standard ogs executable in the standard-path
if ogs_exe is None:
check_ogs = shutil.which(
ogs_name, path=OGS5PY_CONFIG
) or shutil.which(ogs_name)
if check_ogs is None:
print(
"Please put the ogs executable in the default sys path: "
+ str(os.defpath.split(os.pathsep))
)
print("Or provide the path to your executable")
return False
ogs_exe = check_ogs
else:
if sys.platform == "win32" and ogs_exe[-4:] == ".lnk":
print("Don't use file links under windows...")
return False
if os.path.islink(ogs_exe):
ogs_exe = os.path.realpath(ogs_exe)
if os.path.exists(ogs_exe):
if not os.path.isfile(ogs_exe):
ogs_exe = os.path.join(ogs_exe, ogs_name)
else:
print("The given ogs_exe does not exist...")
return False
# use absolute path since we change the cwd in the ogs call
ogs_exe = os.path.abspath(ogs_exe)
# create the command to call ogs
args = [ogs_exe, self.task_id]
# add optional output directory
# check if output directory is an absolute path with os.path.isabs
# otherwise set it in the task_root directory
if self.has_output_dir:
# format the outputdir
output_dir = os.path.normpath(self.output_dir)
# check if outputdir is given as absolut path
if not os.path.isabs(output_dir):
# if not, put the outputfolder in the task_root
output_dir = os.path.join(
os.path.abspath(self.task_root), output_dir
)
# create the outputdir
if not os.path.exists(output_dir):
os.makedirs(output_dir)
# append the outputdir to the ogs-command
args.append("--output-directory")
args.append(output_dir)
else:
output_dir = os.path.abspath(self.task_root)
# prevent eraising files...
if not save_log:
log_name = None
# set standard log_name
if log_name is None:
log_name = (
self.task_id
+ "_"
+ time.strftime("%Y-%m-%d_%H-%M-%S")
+ "_log.txt"
)
# put the logfile in the defined output-dir
if log_path is None:
log_path = output_dir
log = os.path.join(log_path, log_name)
# create a splitted output stream (to file and stdout)
out = Output(log, print_log=print_log)
# call ogs with pexpect
child = CmdRun(
" ".join(args),
timeout=timeout,
logfile=out,
encoding=out.encoding,
cwd=os.path.abspath(self.task_root),
)
# wait for ogs to finish
child.expect(pexpect.EOF)
if sys.platform != "win32":
child.close()
self.exitstatus = child.exitstatus
else:
self.exitstatus = child.wait()
success = self.exitstatus == 0
# close the output stream
out.close()
if not save_log:
os.remove(log)
return success