Mod: improved file and case usage
Mod: openfoam submodule documentation
This commit is contained in:
parent
33ff54bd3b
commit
12ee8ab0b0
@ -1,49 +1,85 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
import os, shutil
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import re
|
||||
from copy import deepcopy
|
||||
import pathlib
|
||||
|
||||
from . import FoamFile
|
||||
|
||||
|
||||
class FoamCase(object):
|
||||
def __init__(self, foamfiles: list = None, path: str = None):
|
||||
def __init__(self, files: FoamFile | list[FoamFile] = None, path: str = None):
|
||||
|
||||
self.path = path or os.path.abspath("")
|
||||
self.path = path
|
||||
self._files = []
|
||||
|
||||
if files is not None:
|
||||
self.add(files)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
content = [ file.object for file in self._files ]
|
||||
|
||||
if foamfiles:
|
||||
self.extend(foamfiles)
|
||||
return "<FoamCase: {}>".format(", ".join(content) or None)
|
||||
|
||||
def add(self, files: FoamFile | list[FoamFile]):
|
||||
if type(files) is not list:
|
||||
assert type(files) is FoamFile, "passed object is not a FoamFile"
|
||||
|
||||
files = [ files ]
|
||||
|
||||
for file in files:
|
||||
assert type(file) is FoamFile, "passed object is not a FoamFile"
|
||||
assert file.object is not None, "FoamFile object attribute is None"
|
||||
|
||||
for n, _file in enumerate(self._files):
|
||||
if _file.object == file.object:
|
||||
self._files.pop(n)
|
||||
self._files.append(file)
|
||||
|
||||
return self
|
||||
|
||||
def __enter__(self):
|
||||
self.__curpath = os.path.abspath("")
|
||||
os.chdir(self.path)
|
||||
return
|
||||
self._files.append(file)
|
||||
|
||||
return self
|
||||
|
||||
def __add__(self, files: FoamFile | list[FoamFile]):
|
||||
return self.add(files)
|
||||
|
||||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||
os.chdir(self.__curpath)
|
||||
self.__curpath = None
|
||||
def write(self, path: str = None):
|
||||
path = pathlib.Path(path or self.path or "")
|
||||
|
||||
def append(self, ff: FoamFile):
|
||||
if FoamFile in ff.__class__.mro():
|
||||
setattr(self, ff.header["object"], deepcopy(ff))
|
||||
for file in self._files:
|
||||
path /= (
|
||||
file.location + "/" + file.object
|
||||
if file.location else file.object
|
||||
)
|
||||
|
||||
else:
|
||||
raise Exception("Trying to put not a FoamFile to FoamCase.")
|
||||
file.write(path.resolve())
|
||||
|
||||
def read(self, path: str = None):
|
||||
path = pathlib.Path(path or self.path or "")
|
||||
|
||||
def extend(self, foamfiles: list):
|
||||
for ff in foamfiles:
|
||||
self.append(ff)
|
||||
for file in self._files:
|
||||
path /= (
|
||||
file.location + "/" + file.object
|
||||
if file.location else file.object
|
||||
)
|
||||
|
||||
def write(self):
|
||||
for value in self.__dict__.values():
|
||||
if FoamFile in value.__class__.mro():
|
||||
value.write(self.path)
|
||||
file.read(path.resolve())
|
||||
|
||||
def read(self):
|
||||
for value in self.__dict__.values():
|
||||
if FoamFile in value.__class__.mro():
|
||||
value.read()
|
||||
def remove(self, path: str = None):
|
||||
path = pathlib.Path(path or self.path or "")
|
||||
|
||||
for file in self._files:
|
||||
path /= (
|
||||
file.location + "/" + file.object
|
||||
if file.location else file.object
|
||||
)
|
||||
|
||||
file.remove(path.resolve())
|
||||
|
||||
def clean(self, included: list = ["0", "constant", "system"]):
|
||||
regxs = [
|
||||
@ -72,5 +108,3 @@ class FoamCase(object):
|
||||
|
||||
if os.path.isfile(file):
|
||||
os.remove(file)
|
||||
|
||||
|
||||
|
@ -1,7 +1,7 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
from typing import List
|
||||
from .runner import FoamRunner
|
||||
from . import FoamRunner
|
||||
|
||||
|
||||
###
|
||||
# meshConversion
|
||||
@ -186,6 +186,7 @@ def simpleFoam(parallel: bool = False, run: bool = True, **kwargs) -> FoamRunner
|
||||
|
||||
return runner
|
||||
|
||||
|
||||
###
|
||||
# postProcessing
|
||||
##
|
||||
|
@ -8,15 +8,18 @@ from . import conversion
|
||||
class FoamFile(object):
|
||||
def __init__(
|
||||
self,
|
||||
filename: str = None,
|
||||
_version: float = 2.0,
|
||||
_format: str = "ascii",
|
||||
_class: str = "dictionary",
|
||||
_object: str = None,
|
||||
_location: str = None,
|
||||
filename: str = None
|
||||
_location: str = None
|
||||
):
|
||||
"""A FoamFile object.
|
||||
|
||||
:param filename:
|
||||
Can be used as shortcut to set _location and _object,
|
||||
_location and _object parameters will be ignored.
|
||||
:param _version:
|
||||
Version of the file format, current is 2.0.
|
||||
:param _format:
|
||||
@ -29,9 +32,6 @@ class FoamFile(object):
|
||||
:param _location:
|
||||
Path to the parent directory of the file according
|
||||
to the case root.
|
||||
:param filename:
|
||||
Can be used as shortcut to set _location and _object,
|
||||
_location and _object parameters will be ignored.
|
||||
"""
|
||||
|
||||
if filename:
|
||||
@ -50,6 +50,26 @@ class FoamFile(object):
|
||||
if _location:
|
||||
self.header["location"] = f'"{ _location }"'
|
||||
|
||||
@property
|
||||
def version(self) -> str:
|
||||
return self.header.get("version")
|
||||
|
||||
@property
|
||||
def format(self) -> str:
|
||||
return self.header.get("format")
|
||||
|
||||
@property
|
||||
def class_(self) -> str:
|
||||
return self.header.get("class")
|
||||
|
||||
@property
|
||||
def object(self) -> str:
|
||||
return self.header.get("object")
|
||||
|
||||
@property
|
||||
def location(self) -> str:
|
||||
return self.header.get("location")
|
||||
|
||||
def __getitem__(self, key):
|
||||
return self.content[key]
|
||||
|
||||
@ -72,16 +92,27 @@ class FoamFile(object):
|
||||
def __repr__(self) -> str:
|
||||
return "<FoamFile: {}>".format(self.header["object"] or None)
|
||||
|
||||
def __add__(self, file):
|
||||
from . import FoamCase
|
||||
|
||||
assert type(file) is FoamFile
|
||||
|
||||
return FoamCase([ self, file ])
|
||||
|
||||
def read(self, filename: str = None):
|
||||
"""Read a FoamFile.
|
||||
|
||||
:param filename:
|
||||
Path to the file. If None, use location from header with
|
||||
current working directory.
|
||||
Path to the file. If None, use location and object from header with
|
||||
the current working directory.
|
||||
:return:
|
||||
Self.
|
||||
"""
|
||||
path = pathlib.Path(filename or self.header["location"]).resolve()
|
||||
filename = (
|
||||
filename or self.location + "/" + self.object
|
||||
if self.location else self.object
|
||||
)
|
||||
path = pathlib.Path(filename).resolve()
|
||||
header, content = conversion.read_foamfile(path)
|
||||
|
||||
self.header = header
|
||||
@ -97,8 +128,28 @@ class FoamFile(object):
|
||||
"""Write a FoamFile to the file.
|
||||
|
||||
:param filename:
|
||||
Path to the file. If None, use location from header with
|
||||
current working directory..
|
||||
Path to the file. If None, use location and object from header with
|
||||
the current working directory.
|
||||
"""
|
||||
filename = pathlib.Path(filename or self.header["location"]).resolve()
|
||||
conversion.write_foamfile(self.header, self.content, filename)
|
||||
filename = (
|
||||
filename or self.location + "/" + self.object
|
||||
if self.location else self.object
|
||||
)
|
||||
path = pathlib.Path(filename).resolve()
|
||||
conversion.write_foamfile(self.header, self.content, path)
|
||||
|
||||
def remove(self, filename: str = None):
|
||||
"""Remove a FoamFile.
|
||||
|
||||
:param filename:
|
||||
Path to the file. If None, use location and object from header with
|
||||
the current working directory.
|
||||
"""
|
||||
filename = (
|
||||
filename or self.location + "/" + self.object
|
||||
if self.location else self.object
|
||||
)
|
||||
path = pathlib.Path(filename).resolve()
|
||||
|
||||
if path.exists():
|
||||
pathlib.os.remove(filename)
|
||||
|
@ -2,14 +2,13 @@
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from typing import List
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class FoamRunner(object):
|
||||
def __init__(self, command: str, args: List[str] = None, mpi: bool = False, cwd: str = None, logpath: str = None, exit: bool = False):
|
||||
def __init__(self, command: str, args: list[str] = None, mpi: bool = False, cwd: str = None, logpath: str = None, exit: bool = False):
|
||||
self.command = command
|
||||
self.args = args
|
||||
self.mpi = mpi
|
||||
@ -20,7 +19,7 @@ class FoamRunner(object):
|
||||
self.error = ""
|
||||
self.returncode = 0
|
||||
|
||||
def fullcommand(self) -> List[str]:
|
||||
def fullcommand(self) -> list[str]:
|
||||
command = []
|
||||
|
||||
if self.mpi:
|
||||
@ -49,8 +48,6 @@ class FoamRunner(object):
|
||||
if self.logpath:
|
||||
with proc, open(self.logpath, "w") as io:
|
||||
while True:
|
||||
|
||||
|
||||
output = proc.stdout.read(1)
|
||||
|
||||
if output == "" and proc.poll() is not None:
|
||||
@ -80,4 +77,4 @@ class FoamRunner(object):
|
||||
if not self.returncode == 0 and self.exit:
|
||||
raise Exception(f"Subprocess failed: { self.error }")
|
||||
|
||||
return self.output, self.error, self.returncode
|
||||
return self.output, self.error, self.returncode
|
||||
|
Loading…
Reference in New Issue
Block a user