"""Boundary condition placeholders. BoCo class definition"""
import os
from typing import Any, Union, IO
if os.environ.get('__USE_CUDA', '0') == '1':
import cupy as np
from cupy import ndarray
else:
import numpy as np
from numpy import ndarray
import re
from g3d.gas import Gas
from g3d.defs import G3DDisplay, G3DEntity
from g3d.defs import BCTypes
from g3d.defs import WALLTypes
from g3d.defs import IOFLOWTypes
from g3d.defs import FMPTypes
from g3d.defs import FunctionByPass
from g3d.mesh.geo import Face
from g3d.flow.vars import Cons
from g3d.flow.vars import Auxv
[docs]
class BoCo(Face):
"""Boundary condition entity BoCo (extends class Face in mesh/geo)"""
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Initialize instance of class BoCo"""
self._cons = Cons()
self._auxv = Auxv()
self._t0 = 0.
self._p0 = 0.
self._defined = False
super().__init__(*args, **kwargs)
def _set_default(self) -> None:
super()._set_default()
self._tag = 'g3d_boco'
def _set_from_args(self, *args: Any, **kwargs: Any) -> ndarray:
bps = self._get_function_evaluate_status(**kwargs)
if bps is FunctionByPass.SELF:
args_stat = np.ones(len(args))
else:
args_stat = super()._set_from_args(*args)
if bps in [FunctionByPass.EVALUATE, FunctionByPass.SELF]:
if np.sum(args_stat) > 0:
for i in range(len(args)):
if args_stat[i] == 1:
arg = args[i]
if type(arg) is BoCo:
args_stat[i] = 0
self.copy(arg)
elif type(arg) is Cons:
args_stat[i] = 0
self._cons.copy(arg)
elif type(arg) is Auxv:
args_stat[i] = 0
self._auxv.copy(arg)
return args_stat
[docs]
def set_from_key_value(self, key: str, value: Any) -> bool:
"""Sets BoCo parameters from provided key => value data
Parameters
----------
key : str :
parameter key/tag
value : Any :
parameter value
Returns
-------
bool
key found status
"""
if super().set_from_key_value(key, value):
return True
elif self._cons.set_from_key_value(key, value):
return True
elif self._auxv.set_from_key_value(key, value):
return True
elif key in ['cons', 'auxv', 'face_def']:
self._set_from_args(value)
return True
elif key == 't0':
self._t0 = value
return True
elif key == 'p0':
self._p0 = value
return True
return False
# class set:ers and get:ers
@property
def cons(self) -> Cons:
"""Cons (conservative variables placeholder)
Returns
-------
Cons
conservative variables placeholder
"""
return self._cons
@cons.setter
def cons(self, val: Cons) -> None:
self._cons = val
@property
def auxv(self) -> Auxv:
"""Auxv (auxiliary (primitive) variables placeholder)
Returns
-------
Auxv
primitive variables placeholder
"""
return self._auxv
@auxv.setter
def auxv(self, val: Auxv) -> None:
self._auxv = val
@property
def t0(self) -> float:
"""To (total temperature [K])
Returns
-------
float
total temperature
"""
return self._t0
@t0.setter
def t0(self, val: float) -> None:
self._t0 = val
@property
def T0(self) -> float:
"""To (total temperature [K])
Returns
-------
float
total temperature
"""
return self._t0
@T0.setter
def T0(self, val: float) -> None:
self._t0 = val
@property
def p0(self) -> float:
"""Po (total pressure [Pa])
Returns
-------
float
total pressure
"""
return self._p0
@p0.setter
def p0(self, val: float) -> None:
self._p0 = val
@property
def defined(self) -> bool:
"""Defined status
Returns
-------
bool
status
"""
return self._defined
@defined.setter
def defined(self, b: bool) -> None:
self._defined = b
# Auxv - set:er/get:er re-directs:
@property
def p(self) -> float:
"""Pressure [Pa]
Returns
-------
float
pressure
"""
return self._auxv.p
@p.setter
def p(self, val: float) -> None:
self._auxv.p = val
@property
def u(self) -> float:
"""Velocity in the x-direction [m/s]
Returns
-------
float
x-velocity
"""
return self._auxv.u
@u.setter
def u(self, val: float) -> None:
self._auxv.u = val
@property
def v(self) -> float:
"""Velocity in the y-direction [m/s]
Returns
-------
float
y-velocity
"""
return self._auxv.v
@v.setter
def v(self, val: float) -> None:
self._auxv.v = val
@property
def w(self) -> float:
"""Velocity in the z-direction [m/s]
Returns
-------
float
z-velocity
"""
return self._auxv.w
@w.setter
def w(self, val: float) -> None:
self._auxv.w = val
@property
def t(self) -> float:
"""Temperature [K]
Returns
-------
float
temperature
"""
return self._auxv.t
@t.setter
def t(self, val: float) -> None:
self._auxv.t = val
@property
def c(self) -> float:
"""Speed of sound [m/s]
Returns
-------
float
speed of sound
"""
return self._auxv.c
@c.setter
def c(self, val: float) -> None:
self._auxv.c = val
# Cons - set:er/get:er re-directs:
@property
def ro(self) -> float:
"""Density [kg/m^3]
Returns
-------
float
density
"""
return self._cons.ro
@ro.setter
def ro(self, val: float) -> None:
self._cons.ro = val
@property
def ru(self) -> float:
"""Momentum in x-direction [kg/(m^2 s)]
Returns
-------
float
x-momentum
"""
return self._cons.ru
@ru.setter
def ru(self, val: float) -> None:
self._cons.ru = val
@property
def rv(self) -> float:
"""Momentum in y-direction [kg/(m^2 s)]
Returns
-------
float
y-momentum
"""
return self._cons.rv
@rv.setter
def rv(self, val: float) -> None:
self._cons.rv = val
@property
def rw(self) -> float:
"""Momentum in z-direction [kg/(m^2 s)]
Returns
-------
float
z-momentum
"""
return self._cons.rw
@rw.setter
def rw(self, val: float) -> None:
self._cons.rw = val
@property
def et(self) -> float:
"""Total energy [kg J / m^3]
Returns
-------
float
total energy
"""
return self._cons.et
@et.setter
def et(self, val: float) -> None:
self._cons.et = val
# class member functions:
[docs]
def allocate_FMP(self, mp: int = 2) -> None:
"""Allocates flux molecule placeholder
Parameters
----------
mp: int :
(Default value = 2)
"""
if self._face_type is BCTypes.SING:
super().allocate_FMP(1)
else:
super().allocate_FMP()
if self._face_type is BCTypes.WALLS and self._it is WALLTypes.VISCOUS_WALL_FUNCTION:
self._FMP.allocate_tmp_arrays(5)
if self._face_type is BCTypes.IOFLOW:
self._FMP.allocate_tmp_arrays(2)
[docs]
def get_status(self) -> bool:
"""Gets boundary condition status
Returns
-------
bool
boundary condition status
"""
return self._defined
[docs]
def update_status(self) -> None:
"""Changes boundary defined status"""
self._defined = not self._defined
[docs]
def get_velocity(self) -> ndarray:
"""Get boundary condition velocity vector
Returns
-------
ndarray
velocity vector
"""
return self._auxv.get_velocity_vector()
[docs]
def copy(self, boco: 'BoCo') -> None:
"""Copies data from boundary condition prototype object
Parameters
----------
boco: BoCo :
prototype object - instance of BoCo
"""
super().copy(boco)
if type(boco) is BoCo: # ensures that call to copy from parent with Face as input works
self._cons.copy(boco.cons)
self._auxv.copy(boco.auxv)
self._t0 = boco.t0
self._p0 = boco.p0
[docs]
def bc_update(self, boco: 'BoCo') -> None:
"""Update boundary condition settings based on settings of provided BoCo
Parameters
----------
boco : BoCo :
prototype - instance of boundary condition object
"""
if boco.face_type is not None:
self._face_type = boco.face_type
if boco.it is not None:
self._it = boco.it
self._cons.copy(boco.cons)
self._auxv.copy(boco.auxv)
self._t0 = boco.t0
self._p0 = boco.p0
[docs]
def bc_update_family(self, bc_type: Any) -> None:
"""Update boundary condition family
Parameters
----------
bc_type : BCTypes :
boundary condition type
valid options: 'UNDEFINED', 'WALLS', 'IOFLOW', 'SYMM', 'SING', 'STREAM_TUBE_WALLS'
"""
if type(bc_type) is BCTypes:
self._face_type = bc_type
[docs]
def bc_update_type(self, bc_it: int) -> None:
"""Updates bc it parameter
Parameters
----------
bc_it : int :
boundary condition identifier
"""
self._it = bc_it
[docs]
def print(self) -> None:
"""Prints boundary condition data (no input)"""
super().print()
self._cons.print()
self._auxv.print()
print(' total pressure %11.5E [Pa] ' % self._p0)
print(' total temperature %11.5E [K] ' % self._t0)
[docs]
def bc_update_speed_of_sound(self, gas: Gas) -> None:
"""Update speed of sound
Parameters
----------
gas : Gas :
gas model
"""
self._auxv.c = gas.get_speed_of_sound(self._auxv.t)
[docs]
def bc_update_primitive(self, gas: Gas) -> None:
"""Update boundary primitive variables (auxiliary variables)
Parameters
----------
gas : Gas :
gas model
"""
self._auxv.set_from_conservative(self._cons.ro, self._cons.ru, self._cons.rv, self._cons.rw, self._cons.et, gas)
[docs]
def bc_update_conservative(self, gas: Gas) -> None:
"""Update boundary conservative variables
Parameters
----------
gas : Gas :
gas model
"""
self._cons.set_from_primitive(self._auxv.p, self._auxv.u, self._auxv.v, self._auxv.w, self._auxv.t, gas)
[docs]
def get_n_ghost_cells(self) -> int:
"""Get boundary ghost cell count
Returns
-------
int
ghost cell count
"""
return self.get_cell_count() * 2
[docs]
def get_n_ghost_nodes(self) -> int:
"""Get boundary ghost node count
Returns
-------
int
ghost node count
"""
return self.get_node_count()
[docs]
def ghost_report(self) -> None:
"""Generates a boundary ghost report"""
print(' surface patch tag: %s' % self._tag)
print(' ghosts cells: %d' % self.get_n_ghost_cells())
print(' ghosts nodes: %d' % self.get_n_ghost_nodes())
[docs]
class BoundaryCondition(G3DEntity):
"""Definition of class BoundaryCondition"""
def __init__(self, *args: Any, **kwargs: Any) -> None:
super().__init__(*args, **kwargs)
self.__size = 0
self._bc_type = BCTypes.UNDEFINED
def _set_default(self) -> None:
self._tag = "boundary"
self.__reset()
def __reset(self) -> None:
self.__a = []
@property
def size(self) -> int:
"""Placeholder size
Returns
-------
int
size
"""
return self.__size
@property
def bc_type(self) -> Any:
"""Boundary condition type
Returns
-------
BCTypes
boundary condition type
"""
return self._bc_type
@bc_type.setter
def bc_type(self, val: Any) -> None:
if type(val) is BCTypes:
self._bc_type = val
[docs]
def allocate_FMPs(self) -> None:
"""Allocates boundary flux molecule placeholders"""
for boco in self.__a:
boco.allocate_FMP()
boco.FMP.type = FMPTypes.BOUNDARY
[docs]
def deallocate_FMPs(self) -> None:
"""De-allocates boundary flux molecule placeholders"""
for boco in self.__a:
boco.deallocate_FMP()
[docs]
def get_max_FMP_size(self) -> int:
"""Get max boundary flux molecule placeholder size
Returns
-------
int
size
"""
FMP_size = 0
for boco in self.__a:
FMP_size = max(FMP_size, boco.FMP_size)
return FMP_size
[docs]
def get_FMP_size(self) -> int:
"""Get boundary accumulated boundary flux molecule placeholder size
Returns
-------
int
size
"""
FMP_size = 0
for boco in self.__a:
FMP_size += boco.FMP_size
return FMP_size
[docs]
def copy(self, bc: 'BoundaryCondition') -> None:
"""Copies data from prototype BoundaryCondition object
Parameters
----------
bc: BoundaryCondition :
instance of BoundaryCondition
"""
self.__reset()
A = bc.get_face_defs()
for a in A:
self.add_face_def(a)
self._tag = bc.tag
self._bc_type = bc.bc_type
[docs]
def get_face_defs(self) -> list:
"""Get boundary face definitions
Returns
-------
list
face definitions
"""
return self.__a
[docs]
def add_face_def(self, a: BoCo) -> None:
"""Add face definition to boundary placeholder
Parameters
----------
a : BoCo :
boundary condition definition
"""
self.__a.append(a)
self.__size = len(self.__a)
[docs]
def read_from_file(self, f: IO[str], n_faces: int, bc_type: Any, bc_tag: str) -> None:
"""Read face definition from file
Parameters
----------
f : IO[str] :
file handle
n_faces : int :
number of faces
bc_type : BCTypes :
boundary type
bc_tag : str :
boundary condition face tag
"""
for i in range(n_faces):
boco = BoCo()
boco.read_def_from_file(f)
boco.tag = bc_tag
boco.face_type = bc_type
if bc_type is BCTypes.IOFLOW:
# falls back to inviscid wall if nothing else is specified
boco.it = IOFLOWTypes.UNDEFINED
if bc_type is BCTypes.WALLS:
# default wall type is viscous resolved
# if running solver in Euler mode, diffusive fluxes are not calculated which
# means that inviscid walls will be used implicitly
boco.it = WALLTypes.VISCOUS_RESOLVED
self.add_face_def(boco)
[docs]
def print(self) -> None:
print('BC Type: %s ' % self._bc_type.name)
print('n_faces: %4d' % self.__size)
for idx in range(self.__size):
print(' face %4d:' % (idx + 1))
self.__a[idx].print()
[docs]
def get_face_count(self, face_type: Any = None) -> int:
"""Get face count for specified face type
Parameters
----------
face_type : BCTypes :
face type (Default value = None)
Returns
-------
int
face count
"""
face_count = 0
for boco in self.__a:
if face_type is None or boco.face_type is face_type:
face_count += boco.get_face_count()
return face_count
[docs]
def get_n_ghost_cells(self) -> int:
"""Get number of allocated ghost cells
Returns
-------
int
ghost cell count
"""
n_ghost_cells = 0
for boco in self.__a:
n_ghost_cells = max(n_ghost_cells, boco.get_n_ghost_cells())
return n_ghost_cells
[docs]
def get_n_ghost_nodes(self) -> int:
"""Get number of allocated ghost nodes
Returns
-------
int
ghost node count
"""
n_ghost_nodes = 0
for boco in self.__a:
n_ghost_nodes = max(n_ghost_nodes, boco.get_n_ghost_nodes())
return n_ghost_nodes
[docs]
def ghost_report(self) -> None:
"""Prints a ghost cell/node report"""
print(' boundary type: %s' % self._bc_type.name)
print(' ghost cells: %d' % self.get_n_ghost_cells())
print(' ghost nodes: %d' % self.get_n_ghost_nodes())
for boco in self.__a:
boco.ghost_report()
[docs]
def update_FMP_metrics(self, x: ndarray, y: ndarray, z: ndarray) -> None:
"""Update boundary flux molecule metrics
Parameters
----------
x : ndarray :
x-coordinate vector
y : ndarray :
y-coordinate vector
z : ndarray :
z-coordinate vector
"""
for boco in self.__a:
boco.FMP.update_metrics(x, y, z)
if boco.face_type.value != BCTypes.STREAM_TUBE_WALLS.value:
if boco.face_type is BCTypes.SYMM or boco.face_type is BCTypes.SING:
if boco.face_type is BCTypes.SYMM:
boco.FMP.update_symmetry_ghost_nodes(x, y, z)
boco.FMP.update_diff_metrics(x, y, z)
else:
boco.FMP.update_diff_metrics(x, y, z, boco.get_face_count())
if boco.face_type is BCTypes.WALLS:
boco.FMP.update_wall_diff_metrics(x, y, z)
[docs]
def check_FMP_metrics(self, x: ndarray, y: ndarray, z: ndarray) -> None:
"""Check boundary flux molecule metrics
Parameters
----------
x : ndarray :
x-coordinate vector
y : ndarray :
y-coordinate vector
z : ndarray :
z-coordinate vector
"""
for boco in self.__a:
boco.FMP.check_metrics(x, y, z)
[docs]
def check_FMP_cell_metrics(self, ic: int, check_sum: Union[list, ndarray]) -> None:
"""Check boundary flux molecule cell metrics
Parameters
----------
ic : int :
cell index
check_sum : ndarray or list:
check sum array
"""
print(' face type: %s' % self._bc_type.name)
for boco in self.__a:
boco.check_FMP_cell_metrics(ic, check_sum)
[docs]
def print_FMP_data(self, face_type: Any = BCTypes.UNDEFINED, face_tag: str = 'all', f: Any = None) -> None:
"""Prints boundary flux molecule data
Parameters
----------
face_type : BCTypes :
boundary condition type (Default value = BCTypes.UNDEFINED)
face_tag : str :
boundary condition face tag (Default value = 'all')
f : Any :
file handle (Default value = None)
"""
if face_type is self._bc_type or face_type is BCTypes.UNDEFINED:
if f is not None:
f.write('Face type: %s\n' % self._bc_type.name)
else:
print('Face type: %s\n' % self._bc_type.name)
for boco in self.__a:
print(face_tag, boco.tag)
if boco.tag == face_tag or face_tag == 'all':
if f is not None:
f.write('Face tag: %s\n' % boco.tag)
else:
print('Face tag: %s' % boco.tag)
boco.FMP.print_data(f)
[docs]
class BoundaryConditionPlaceHolder(G3DEntity):
"""Definition of class BoundaryConditionPlaceHolder"""
def __init__(self, *args: Any, **kwargs: Any) -> None:
super().__init__(*args, **kwargs)
self.__size = 0
self.__boundaries = []
[docs]
def allocate_placeholders(self) -> None:
"""Allocate placeholder data containers"""
for boundary in self.__boundaries:
boundary.allocate_FMPs()
[docs]
def get_boundary_array(self) -> list:
"""Get boundary array
Returns
-------
list
boundary array
"""
return self.__boundaries
[docs]
def get_boundary(self, t: Any) -> Union[BoundaryCondition, None]:
"""Get boundary condition
Parameters
----------
t : BCTypes :
boundary condition type
Returns
-------
BoundaryCondition
boundary
"""
if type(t) is BCTypes:
for boundary in self.__boundaries:
if boundary.bc_type == t:
return boundary
return self.add_boundary(t)
return None
[docs]
def add_boundary(self, t: Any) -> Union[BoundaryCondition, None]:
"""Add boundary container
Parameters
----------
t : BCTypes :
boundary condition type
Returns
-------
BoundaryCondition
boundary
"""
if type(t) is BCTypes:
boundary = BoundaryCondition()
boundary.bc_type = t
self.__boundaries.append(boundary)
self.__size = len(self.__boundaries)
return boundary
return None
[docs]
def add_Q1D_boundary_definitions(self, NI: int, left_end_BC: Any = BCTypes.IOFLOW,
right_end_BC: Any = BCTypes.IOFLOW) -> None:
"""Add quasi-one-dimensional solver boundary definitions
Parameters
----------
NI : int :
node count
left_end_BC : BCTypes :
left-end boundary condition (Default value = BCTypes.IOFLOW)
right_end_BC : BCTypes :
right-end boundary condition (Default value = BCTypes.IOFLOW)
"""
if type(left_end_BC) is BCTypes and type(right_end_BC) is BCTypes:
boundary = self.get_boundary(left_end_BC)
left = BoCo()
left.set_def([1, 1, 1, 1, 5, 3, 2, 2])
left.tag = 'left'
left.face_type = left_end_BC
boundary.add_face_def(left)
boundary = self.get_boundary(right_end_BC)
right = BoCo()
right.set_def([1, NI, 1, 1, 3, 5, 2, 2])
right.tag = 'right'
right.face_type = right_end_BC
boundary.add_face_def(right)
[docs]
def read_from_file(self, f: IO[str], bc_type: Any) -> None:
"""Read boundary definition from file
Parameters
----------
f : IO[str] :
file handle
bc_type : BCTypes :
boundary condition type
"""
if type(bc_type) is BCTypes:
regex_tag = re.compile(r'^\s*(?P<tag>[\w\-.]+)')
regex_n_faces = re.compile(r'^\s*(?P<n_faces>\d+)\s*$')
regex_empty_line = re.compile(r'^\s*$')
boundary = self.get_boundary(bc_type)
bc_tag = bc_type.name.lower()
n_faces = 1
line = f.readline()
while regex_empty_line.search(line):
line = f.readline()
match = regex_n_faces.search(line)
if match:
n_faces = int(match.group('n_faces'))
else:
match = regex_tag.search(line)
if match:
bc_tag = match.group('tag')
tmp = f.tell()
line = f.readline()
while regex_empty_line.search(line):
tmp = f.tell()
line = f.readline()
match = regex_n_faces.search(line)
if match:
n_faces = int(match.group('n_faces'))
else:
f.seek(tmp)
boundary.read_from_file(f, n_faces, bc_type, bc_tag)
[docs]
def print(self) -> None:
g3d_disp = G3DDisplay()
if g3d_disp.verbose:
if self.__size:
print(' number of defined boundary-face groups: %4d' % self.__size)
for boundary in self.__boundaries:
boundary.print()
[docs]
def get_face_count(self, face_type: Any = None) -> int:
"""Extract boundary face count for specified boundary type or all boundaries
Parameters
----------
face_type : BCTypes :
boundary condition type (Default value = None)
Returns
-------
int
face count
"""
face_count = 0
for boundary in self.__boundaries:
face_count += boundary.get_face_count(face_type)
return face_count
[docs]
def get_max_ghost_cells(self) -> int:
"""Finds maximum number of ghost cells in a boundary container
Returns
-------
int
ghost cell count
"""
n_ghost_cells = 0
for boundary in self.__boundaries:
n_ghost_cells = max(n_ghost_cells, boundary.get_n_ghost_cells())
return n_ghost_cells
[docs]
def get_max_ghost_nodes(self) -> int:
"""Finds maximum number of ghost nodes in a boundary container
Returns
-------
int
ghost node count
"""
n_ghost_nodes = 0
for boundary in self.__boundaries:
n_ghost_nodes = max(n_ghost_nodes, boundary.get_n_ghost_nodes())
return n_ghost_nodes
[docs]
def ghost_report(self) -> None:
"""Extracts boundary ghost cell/node report"""
for boundary in self.__boundaries:
boundary.ghost_report()
print('max ghost cells: %d' % self.get_max_ghost_cells())
print('max ghost nodes: %d' % self.get_max_ghost_nodes())
[docs]
def get_max_FMP_size(self) -> int:
"""Get maximum flux molecule placeholder size
Returns
-------
int
size
"""
MaxFMPSize = 0
for boundary in self.__boundaries:
MaxFMPSize = max(MaxFMPSize, boundary.get_max_FMP_size())
return MaxFMPSize
[docs]
def update_FMP_metrics(self, x: ndarray, y: ndarray, z: ndarray) -> None:
"""Updates flux molecule metrics for stored boundary condition containers
Parameters
----------
x : ndarray :
x-coordinate vector
y : ndarray :
y-coordinate vector
z : ndarray :
z-coordinate vector
"""
for boundary in self.__boundaries:
boundary.update_FMP_metrics(x, y, z)
[docs]
def check_FMP_metrics(self, x: ndarray, y: ndarray, z: ndarray) -> None:
"""Checks flux molecule metrics for stored boundary condition containers
Parameters
----------
x : ndarray :
x-coordinate vector
y : ndarray :
y-coordinate vector
z : ndarray :
z-coordinate vector
"""
for boundary in self.__boundaries:
boundary.check_FMP_metrics(x, y, z)
[docs]
def check_FMP_cell_metrics(self, ic: int, check_sum: Union[list, ndarray]) -> None:
"""Checks flux molecule cell metrics
Parameters
----------
ic : int :
cell index
check_sum : ndarray or list :
check sum array
"""
if len(self.__boundaries) != 0:
print(' boundary faces:')
for boundary in self.__boundaries:
boundary.check_FMP_cell_metrics(ic, check_sum)
[docs]
def print_FMP_data(self, face_type: Any = BCTypes.UNDEFINED, face_tag: str = 'all', f: Any = None) -> None:
"""Prints flux molecule data
Parameters
----------
face_type : BCTypes :
boundary condition type (Default value = BCTypes.UNDEFINED)
face_tag : str :
boundary face tag (Default value = 'all')
f : Any :
file handle (Default value = None)
"""
if type(face_type) is BCTypes:
for boundary in self.__boundaries:
boundary.print_FMP_data(face_type, face_tag, f)