"""Connectivity placeholders. Intf class definition."""
import re
import os
from typing import Any, Union, Tuple, 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
from g3d.defs import G3DDisplay, G3DEntity
from g3d.defs import FMPTypes
from g3d.defs import INTFTypes
from g3d.mesh.geo import Face
[docs]
class Intf(Face):
"""Definition of class Intf, extends class Face"""
def __init__(self, *args: Any, **kwargs: Any) -> None:
self._dx = 0.
self._dy = 0.
self._dz = 0.
self._alfa = 0.
super().__init__(*args, **kwargs)
def _set_default(self) -> None:
super()._set_default()
self._tag = 'g3d_interface'
self.__allocate_ghosts = False
@property
def dx(self) -> float:
"""Spatial separation in x-direction (for translational periodic boundaries)
Returns
-------
float
dx
"""
return self._dx
@dx.setter
def dx(self, val: float) -> None:
self._dx = val
@property
def dy(self) -> float:
"""Spatial separation in y-direction (for translational periodic boundaries)
Returns
-------
float
dy
"""
return self._dy
@dy.setter
def dy(self, val: float) -> None:
self._dy = val
@property
def dz(self) -> float:
"""Spatial separation in z-direction (for translational periodic boundaries)
Returns
-------
float
dz
"""
return self._dz
@dz.setter
def dz(self, val: float) -> None:
self._dz = val
@property
def alfa(self) -> float:
"""Angular separation, rotation about x-axis (for rotational periodic boundaries)
Returns
-------
float
alfa
"""
return self._alfa
@alfa.setter
def alfa(self, val: float) -> None:
self._alfa = val
[docs]
def allocate_FMP(self, mp: int = 2) -> None:
"""Allocates flux molecule placeholders
Parameters
----------
mp: int :
(Default value = 2)
"""
super().allocate_FMP(3)
[docs]
def copy(self, intf: 'Intf'):
"""Copies data from prototype Intf object
Parameters
----------
intf: Intf
prototype Intf object
"""
super().copy(intf)
self._dx = intf.dx
self._dy = intf.dy
self._dz = intf.dz
self._alfa = intf.alfa
[docs]
def get_face_delta(self) -> ndarray:
"""Get face spatial separation in three coordinate directions (for translational periodic faces)
Returns
-------
ndarray
[dx, dy, dz]
"""
return np.array([self._dx, self._dy, self._dz])
[docs]
def set_face_delta(self, d: Union[ndarray, list]) -> None:
"""Get face spatial separation in three coordinate directions (for translational periodic faces)
Parameters
----------
d : ndarray
spatial separation in three coordinates [dx, dy, dz]
"""
self._dx = d[0]
self._dy = d[1]
self._dz = d[2]
[docs]
def get_n_ghost_cells(self) -> int:
"""Get number of ghost cells for interface
Returns
-------
int
number of ghost cells
"""
if self.__allocate_ghosts:
return self.get_cell_count() * 2
return 0
[docs]
def get_n_ghost_nodes(self) -> int:
"""Get number of ghost nodes for interface
Returns
-------
int
number of ghost nodes
"""
if self.__allocate_ghosts:
return self.get_node_count()
return 0
[docs]
def ghost_report(self) -> None:
"""Extracts interface 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 Interface(G3DEntity):
"""Definition of class Interface"""
def __init__(self, *args: Any, **kwargs: Any) -> None:
super().__init__(*args, **kwargs)
self.__size = 0
self._intf_type = INTFTypes.UNDEFINED
def _set_default(self) -> None:
self._tag = "interface"
self.__reset()
def __reset(self):
self.__size = 0
self.__a = []
self.__b = []
@property
def size(self) -> int:
"""Placeholder size
Returns
-------
int
size
"""
return self.__size
@property
def intf_type(self) -> Any:
"""Connectivity interface type
Returns
-------
INTFTypes
interface type
"""
return self._intf_type
@intf_type.setter
def intf_type(self, t: Any) -> None:
if type(t) is INTFTypes:
self._intf_type = t
[docs]
def allocate_FMPs(self) -> None:
"""Allocates interface flux molecule placeholders"""
for intf in self.__a:
intf.allocate_FMP()
intf.FMP.type = FMPTypes.BLOCK_INTERFACE
[docs]
def deallocate_FMPs(self) -> None:
"""De-allocates interface flux molecule placeholders"""
for intf in self.__a:
intf.deallocate_FMP()
[docs]
def get_max_FMP_size(self) -> int:
"""Get max interface flux molecule placeholder size
Returns
-------
int
size
"""
FMP_size = 0
for intf in self.__a:
FMP_size = max(FMP_size, intf.FMP_size)
return FMP_size
[docs]
def get_FMP_size(self) -> int:
"""Get interface accumulated boundary flux molecule placeholder size
Returns
-------
int
size
"""
FMP_size = 0
for intf in self.__a:
FMP_size += intf.FMP_size
return FMP_size
[docs]
def copy(self, intf: 'Interface') -> None:
"""Copies data from prototype Interface object
Parameters
----------
intf: Interface
prototype Interface object
"""
self.__reset()
A, B = intf.get_face_defs()
for i in range(len(A)):
self.add_face_def(A[i], B[i])
self._tag = intf.tag
self._intf_type = intf.intf_type
[docs]
def get_face_defs(self) -> Tuple[list, list]:
"""Get interface face definitions
Returns
-------
list
face definitions
"""
return self.__a, self.__b
[docs]
def add_face_def(self, a: Intf, b: Intf):
"""Add face definition to boundary placeholder
Parameters
----------
a: Intf :
face definition A-side
b: Intf :
face definition B-side
"""
self.__a.append(a)
self.__b.append(b)
self.__size = len(self.__a)
[docs]
def read_from_file(self, f: IO[str], n_faces: int, intf_type: Any, intf_tag: str) -> None:
"""Read face definition from file
Parameters
----------
f : file
file handle
n_faces : int
number of faces
intf_type : INTFTypes
interface type
intf_tag : str
interface tag
"""
if type(intf_type) is INTFTypes:
for i in range(n_faces):
intfA = Intf()
intfB = Intf()
intfA.read_def_from_file(f)
intfB.read_def_from_file(f)
intfA.tag = intf_tag
intfB.tag = intf_tag
intfA.face_type = intf_type
intfB.face_type = intf_type
self.add_face_def(intfA, intfB)
[docs]
def print(self) -> None:
print('INTF Type: %s ' % self._intf_type.name)
print('n_faces : %4d' % self.__size)
for idx in range(self.__size):
print(' face %4d:' % (idx + 1))
print(' A-side:')
self.__a[idx].print()
print(' B-side:')
self.__b[idx].print()
[docs]
def get_face_count(self) -> int:
"""Get face count
Returns
-------
int
face count
"""
face_count = 0
for a in self.__a:
face_count += a.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 intf in self.__a:
n_ghost_cells = max(n_ghost_cells, intf.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 intf in self.__a:
n_ghost_nodes = max(n_ghost_nodes, intf.get_n_ghost_nodes())
return n_ghost_nodes
[docs]
def ghost_report(self) -> None:
"""Prints a ghost cell/node report"""
print(' interface type: %s' % self._intf_type.name)
print(' ghost cells: %d' % self.get_n_ghost_cells())
print(' ghost nodes: %d' % self.get_n_ghost_nodes())
for intf in self.__a:
intf.ghost_report()
[docs]
def update_FMP_metrics(self, x: ndarray, y: ndarray, z: ndarray) -> None:
"""Update interface flux molecule metrics
Parameters
----------
x: ndarray :
x coordinate vector
y: ndarray :
y coordinate vector
z: ndarray :
z coordinate vector
"""
for intf in self.__a:
intf.FMP.update_metrics(x, y, z)
intf.FMP.update_diff_metrics(x, y, z)
[docs]
def check_FMP_metrics(self, x: ndarray, y: ndarray, z: ndarray) -> None:
"""Check interface flux molecule metrics
Parameters
----------
x: ndarray :
x coordinate vector
y: ndarray :
y coordinate vector
z: ndarray :
z coordinate vector
"""
for intf in self.__a:
intf.FMP.check_metrics(x, y, z)
[docs]
def check_FMP_cell_metrics(self, ic: int, check_sum: Union[list, ndarray]) -> None:
"""Check interface flux molecule cell metrics
Parameters
----------
ic : int :
cell index
check_sum : ndarray :
check sum array
"""
print(' face type: %s' % self._intf_type.name)
for intf in self.__a:
intf.check_FMP_cell_metrics(ic, check_sum)
[docs]
def print_FMP_data(self, face_type: Any = INTFTypes.UNDEFINED, face_tag: str = 'all', f: IO[str] = None) -> None:
"""Prints interface flux molecule data
Parameters
----------
face_type : INTFTypes :
interface type (Default value = INTFTypes.UNDEFINED)
face_tag : str :
interface tag (Default value = 'all')
f : file :
file handle (Default value = None)
"""
if face_type is self._intf_type or face_type is INTFTypes.UNDEFINED:
if f is not None:
f.write('Face type: %s\n' % self._intf_type.name)
else:
print('Face type: %s\n' % self._intf_type.name)
for intf in self.__a:
if intf.tag == face_tag or face_tag == 'all':
if f is not None:
f.write('Face tag: %s\n' % intf.tag)
else:
print('Face tag: %s' % intf.tag)
intf.FMP.print_data(f)
[docs]
class InterfacePlaceHolder(G3DEntity):
"""Definition of class InterfacePlaceHolder"""
def __init__(self, *args: Any, **kwargs: Any) -> None:
super().__init__(*args, **kwargs)
self.__size = 0
self.__interfaces = []
[docs]
def allocate_placeholders(self) -> None:
"""Allocate placeholder data containers"""
for interface in self.__interfaces:
interface.allocate_FMPs()
[docs]
def get_interface_array(self) -> list:
"""Get interface array
Returns
-------
list
interface array
"""
return self.__interfaces
[docs]
def get_interface(self, t: Any) -> Union[Interface, None]:
"""Get interface
Parameters
----------
t : INTFTypes :
interface type
Returns
-------
Interface or None
interface
"""
if type(t) is INTFTypes:
for interface in self.__interfaces:
if interface.intf_type == t:
return interface
return self.add_interface(t)
return None
[docs]
def add_interface(self, t: Any) -> Union[Interface, None]:
"""Add interface container
Parameters
----------
t : INTFTypes :
interface type
Returns
-------
Interface or None
interface
"""
if type(t) is INTFTypes:
interface = Interface()
interface.intf_type = t
self.__interfaces.append(interface)
self.__size = len(self.__interfaces)
return interface
return None
[docs]
def read_from_file(self, f: IO[str], intf_type: Any) -> None:
"""Read interface definition from file
Parameters
----------
f : file :
file handle
intf_type : INTFTypes :
interface type
"""
if type(intf_type) is INTFTypes:
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*$')
interface = self.get_interface(intf_type)
intf_tag = intf_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:
intf_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)
interface.read_from_file(f, n_faces, intf_type, intf_tag)
[docs]
def print(self) -> None:
g3d_disp = G3DDisplay()
if g3d_disp.verbose:
if self.__size:
print(' number of defined connectivity-face groups: %4d' % self.__size)
for interface in self.__interfaces:
interface.print()
[docs]
def get_face_count(self) -> int:
"""Extract interface face count
Returns
-------
int
face count
"""
face_count = 0
for interface in self.__interfaces:
face_count += interface.get_face_count()
return face_count
[docs]
def get_max_ghost_cells(self) -> int:
"""Finds maximum number of ghost cells in an interface container
Returns
-------
int
ghost cell count
"""
n_ghost_cells = 0
for interface in self.__interfaces:
n_ghost_cells = max(n_ghost_cells, interface.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 interface in self.__interfaces:
n_ghost_nodes = max(n_ghost_nodes, interface.get_n_ghost_nodes())
return n_ghost_nodes
[docs]
def ghost_report(self) -> None:
"""Extracts boundary ghost cell/node report"""
for interface in self.__interfaces:
interface.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 interface in self.__interfaces:
MaxFMPSize = max(MaxFMPSize, interface.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 interface containers
Parameters
----------
x: ndarray :
x coordinate vector
y: ndarray :
y coordinate vector
z: ndarray :
z coordinate vector
"""
for interface in self.__interfaces:
interface.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 interface containers
Parameters
----------
x: ndarray :
x coordinate vector
y: ndarray :
y coordinate vector
z: ndarray :
z coordinate vector
"""
for interface in self.__interfaces:
interface.check_FMP_metrics(x, y, z)
[docs]
def check_FMP_cell_metrics(self, ic: int, check_sum: ndarray) -> None:
"""Checks flux molecule cell metrics
Parameters
----------
ic : int :
cell index
check_sum : ndarray :
check sum array
"""
if len(self.__interfaces) != 0:
print(' connectivity faces:')
for interface in self.__interfaces:
interface.check_FMP_cell_metrics(ic, check_sum)
[docs]
def print_FMP_data(self, face_type=INTFTypes.UNDEFINED, face_tag='all', f=None) -> None:
"""Prints flux molecule data
Parameters
----------
face_type : INTFTypes :
interface type (Default value = INTFTypes.UNDEFINED)
face_tag : str :
interface tag (Default value = 'all')
f : file
file handle (Default value = None)
"""
for interface in self.__interfaces:
interface.print_FMP_data(face_type, face_tag, f)