import pytest
import os
from dataclasses import dataclass, field
from PyFoam.RunDictionary.ParsedParameterFile import ParsedParameterFile
from typing import List, Tuple, Optional, Dict, Any
from shutil import copyfile
[docs]def base_dir() -> str:
"""directory of curren test
Returns:
str: directory path of the test
"""
f_name = os.getenv("PYTEST_CURRENT_TEST").split("::")[0]
dir_name = os.path.dirname(f_name)
return dir_name
[docs]def path_log(app_name: str = "") -> str:
"""path of the log file
reads controlDict to get application if app_name not specified
Args:
app_name (str, optional): name of the application. Defaults to read controlDict to get application.
Returns:
str: path to the log file
"""
dir_name = base_dir()
if app_name:
return os.path.join(dir_name, "log." + app_name)
controlDict = os.path.join(dir_name, "system/controlDict")
p = Pyfoam_parser(controlDict)
app_name = p.value("application")
return os.path.join(dir_name, "log." + app_name)
[docs]class Parser:
"""abstract parser class
"""
def __init__(self, filename: str):
self.filename = filename
[docs] def value(self, keyword: str):
pass
[docs] def set(self, keyword: str):
pass
[docs] def writeFile(self):
pass
[docs]class Pyfoam_parser(Parser):
"""pyfoam based openfoam dict parser that modifes a file
Args:
Parser ([type]): abstract class
"""
def __init__(self, filename: str):
self.filename = filename
self._ppp = ParsedParameterFile(self.filename)
def _nested_get(self, dic: Dict, keyword: str):
key_list = keyword.split("/")
key_list[:] = [x for x in key_list if x]
if len(key_list) == 1:
return dic[key_list[0]]
for key in key_list:
dic = dic[key]
return dic
def _nested_set(self, dic: Dict, keyword: str, value: Any):
key_list = keyword.split("/")
key_list[:] = [x for x in key_list if x]
if len(key_list) == 1:
dic[key_list[-1]] = value
return
for key in key_list[:-1]:
dic = dic.setdefault(key, {})
dic[key_list[-1]] = value
[docs] def value(self, keyword: str):
"""get value of key word
Args:
keyword (str): keyword as string e.g. application
or in case of a nested dictionary dict1/subDict1/keyword1
Returns:
[type]: return value
"""
return self._nested_get(self._ppp.content, keyword)
[docs] def set(self, keyword: str, value: Any):
"""set value
Args:
keyword (str): keyword as string e.g. application
or in case of a nested dictionary dict1/subDict1/keyword1
value (Any): new value
"""
self._nested_set(self._ppp.content, keyword, value)
[docs] def writeFile(self):
self._ppp.writeFile()
[docs]class Case_modifiers:
"""modifes and openfoam case by modifying the case files
Args:
case_modifiers (Dict): dict format filename : list of (keyword , value)
e.g.
{
"system/controlDict": [ ("stopAt","writeNow"),
("endTime",10.1) ],
"constant/transportProperties": [ ("water/transportModel","Newtonian"),
("air/transportModel","Newtonian") ]
}
subdicts are seperated by /
dir_name (str): dir of openfoam case
meta_data (Optional[Dict], optional): stores additional information. Defaults to {}.
"""
def __init__(
self, case_modifiers: Dict, dir_name: str, meta_data: Optional[Dict] = {}
):
self.modifiers = case_modifiers
self.dir_name = dir_name
self.meta_data = meta_data
if "script" not in self.meta_data:
self.meta_data["script"] = "Allrun -test"
def __str__(self):
out = str(self.modifiers)
if self.meta_data:
out += str(self.meta_data)
return out
[docs] def add_mod(self, file_path: str, key:str, val: Any):
"""add new file modification
Args:
file_path (str): path to file
key (str): keyword
val (Any): value
"""
if file_path not in self.modifiers:
self.modifiers[file_path] = []
self.modifiers[file_path].append((key, val))
[docs] def update_case(self):
"""
update the based on the specified modifiers
"""
for key in self.modifiers:
bkp_file = key + ".orig"
bkp_path = os.path.join(self.dir_name, bkp_file)
file_path = os.path.join(self.dir_name, key)
# backup file
if not os.path.isfile(bkp_path):
copyfile(file_path, bkp_path)
p = Pyfoam_parser(os.path.join(self.dir_name, key))
for key_val in self.modifiers[key]:
p.set(key_val[0], key_val[1])
p.writeFile()
[docs] def revert_change(self):
"""
revert changes
"""
for key in self.modifiers:
bkp_file = key + ".orig"
bkp_path = os.path.join(self.dir_name, bkp_file)
file_path = os.path.join(self.dir_name, key)
if os.path.isfile(bkp_path):
copyfile(bkp_path, file_path)
else:
os.remove(bkp_path)
[docs]def check_type(c_mod) -> Case_modifiers:
if not isinstance(c_mod, Case_modifiers):
try:
# can also be a tuple of length of
if len(c_mod) == 1: # enables latter extension to multiple parameters
c_mod = c_mod[0]
else:
TypeError("parameter needs to be a Case_modifiers not a tuple")
except:
raise TypeError("parameter needs to be a Case_modifiers")
return c_mod
[docs]@pytest.fixture(scope="class")
def run_case(request):
"""fixture that runs case by exectuting a bash script
The case can be modified by passing the Case_modifiers class
Default name is script name Allrun can be modified by storing the script name
in the meta_data of the Case_modifiers:
c_mod.meta_data["script"] = "SomeScriptName"
Yields:
[Case_modifiers]: Case_modifiers information
"""
mod_case = hasattr(request, "param")
dir_name = base_dir()
c_mod = Case_modifiers({}, dir_name)
if mod_case:
c_mod = check_type(request.param)
nsteps = request.config.getoption("--writeNSteps")
if nsteps:
c_mod.add_mod("system/controlDict", "startFrom", "latestTime")
c_mod.add_mod("system/controlDict", "stopAt", "nextWrite")
c_mod.add_mod("system/controlDict", "writeControl", "timeStep")
c_mod.add_mod("system/controlDict", "writeInterval", nsteps)
c_mod.update_case()
if c_mod.meta_data:
if "script" not in c_mod.meta_data:
c_mod.meta_data["script"] = "Allrun -test"
os.system(f"{dir_name}/{c_mod.meta_data['script']}")
yield c_mod
[docs]@pytest.fixture(scope="class")
def run_reset_case(request):
"""fixture that runs case by exectuting a bash script and reset the case by calling Allclean
The case can be modified by passing the Case_modifiers class
Default name is script name Allrun can be modified by storing the script name
in the meta_data of the Case_modifiers:
c_mod.meta_data["script"] = "SomeScriptName"
Yields:
[Case_modifiers]: Case_modifiers information
"""
mod_case = hasattr(request, "param")
dir_name = base_dir()
c_mod = Case_modifiers({}, dir_name)
if mod_case:
c_mod = check_type(request.param)
nsteps = request.config.getoption("--writeNSteps")
if nsteps:
c_mod.add_mod("system/controlDict", "startFrom", "latestTime")
c_mod.add_mod("system/controlDict", "stopAt", "nextWrite")
c_mod.add_mod("system/controlDict", "writeControl", "timeStep")
c_mod.add_mod("system/controlDict", "writeInterval", nsteps)
c_mod.update_case()
if c_mod.meta_data:
if "script" not in c_mod.meta_data:
c_mod.meta_data["script"] = "Allrun -test"
os.system(f"{dir_name}/{c_mod.meta_data['script']}")
yield c_mod
c_mod.revert_change()
if request.config.getoption("--no-clean-up"):
os.system(f"{dir_name}/Allclean")
[docs]@pytest.fixture(scope="class")
def modify_case(request):
"""modifies the case without running it
The case can be modified by passing the Case_modifiers class
Yields:
[Case_modifiers]: Case_modifiers information
"""
mod_case = hasattr(request, "param")
dir_name = base_dir()
c_mod = Case_modifiers({}, dir_name)
if mod_case:
c_mod = check_type(request.param)
nsteps = request.config.getoption("--writeNSteps")
if nsteps:
c_mod.add_mod("system/controlDict", "startFrom", "latestTime")
c_mod.add_mod("system/controlDict", "stopAt", "nextWrite")
c_mod.add_mod("system/controlDict", "writeControl", "timeStep")
c_mod.add_mod("system/controlDict", "writeInterval", nsteps)
c_mod.update_case()
yield c_mod
c_mod.revert_change()
[docs]@pytest.fixture(scope="class")
def clean_case(request):
"""cleans case by running Allcean
Yields:
[type]: case modifier
"""
mod_case = hasattr(request, "param")
dir_name = base_dir()
c_mod = Case_modifiers({}, dir_name)
if mod_case:
c_mod = check_type(request.param)
c_mod.revert_change()
if request.config.getoption("--no-clean-up"):
os.system(f"{dir_name}/Allclean")
yield c_mod