"""Generic flow field data placeholder (FlowDataPlaceHolder class definition)"""
from typing import Any, Union, Tuple, BinaryIO
import os
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 struct
from g3d.defs import G3DDisplay
from g3d.defs import G3DEntity
from g3d.defs import FunctionByPass
[docs]
class FlowDataPlaceHolder(G3DEntity):
"""Definition of class FlowDataPlaceHolder: a generic placeholder for field data"""
def __init__(self, *args: Any, **kwargs: Any) -> None:
self._size = 0
self._n_cells = 0
self._n_ghosts = 0
self._first = 0
self._last = 0
self.__backup_data = False
self.__n_bak = self._size
self._pointers = []
self._keys = []
self._n_var = 0
super().__init__(*args, **kwargs)
self.setup(**kwargs)
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:
size_set = False
for i in range(len(args)):
if args_stat[i] == 1:
arg = args[i]
if type(arg) is FlowDataPlaceHolder and not size_set:
args_stat[i] = 0
self.copy(arg)
size_set = True
elif type(arg) is int and not size_set:
args_stat[i] = 0
self._size = arg
size_set = True
return args_stat
def _set_default(self) -> None:
super()._set_default()
self._tag = 'g3d_flow_data_placeholder'
self.__reset_sizes()
self._define_variables()
self.__reset_backup()
def __reset_sizes(self) -> None:
self._size = 0
self._n_cells = 0
self._n_ghosts = 0
self._first = 0
self._last = 0
def __update_size(self) -> None:
self.__n_bak = self._size
self._size = self._n_cells + self._n_ghosts
def __update_range(self) -> None:
self._last = self._first + self._n_cells
[docs]
def set_from_key_value(self, key: str, value: Any) -> bool:
"""Sets data from key-value pair input
Parameters
----------
key: str :
keyword
valid class-specific keywords: 'size', 'n_cells', 'first', 'last', 'n_ghosts', 'backup_data'
value: Any :
value corresponding to keyword
Returns
-------
bool
status - True if key is found in list of valid keywords
"""
if super().set_from_key_value(key, value):
return True
elif key == 'size':
self.__n_bak = self._size
self._size = value
if self._size == 0:
self.__reset_sizes()
return True
elif key == 'n_cells':
self._n_cells = value
self.__update_size()
self.__update_range()
return True
elif key == 'last':
self._last = value
return True
elif key == 'first':
self._first = value
self.__update_range()
return True
elif key == 'n_ghosts':
self._n_ghosts = value
self.__update_size()
self.__update_range()
return True
elif key == 'backup_data':
self.__backup_data = value
return True
return False
def __reset_backup(self) -> None:
self.__backup_data = False
for var in self._variables.keys():
self._variables[var]['bak'] = []
[docs]
def backup(self) -> None:
"""Copies data to backup arrays"""
if self.__n_bak > 0 and self.__backup_data:
ll = np.arange(self.__n_bak)
for var in self._variables.keys():
self._variables[var]['bak'] = np.zeros(self.__n_bak)
self._variables[var]['bak'][ll] = self._variables[var]['ref'][ll]
[docs]
def get_from_backup(self, reset_bak: bool = True) -> None:
"""Restores data from backup arrays
Parameters
----------
reset_bak: bool :
(Default value = True)
"""
if self.__n_bak > 0 and self.__backup_data:
ll = np.arange(self.__n_bak)
for var in self._variables.keys():
self._variables[var]['ref'][ll] = self._variables[var]['bak'][ll]
if reset_bak:
self.__reset_backup()
@property
def n_var(self) -> int:
"""Number of variables
Returns
-------
int
variable count
"""
return len(self._pointers)
@property
def first(self) -> int:
"""First index
Returns
-------
int
first index value
"""
return self._first
@first.setter
def first(self, val: int) -> None:
self._first = val
@property
def last(self) -> int:
"""Last index
Returns
-------
int
last index value
"""
return self._last
@last.setter
def last(self, val: int) -> None:
self._last = val
@property
def n_cells(self) -> int:
"""Number of cells
Returns
-------
int
cell count
"""
return self._n_cells
@n_cells.setter
def n_cells(self, val: int) -> None:
self._n_cells = val
@property
def n_ghosts(self) -> int:
"""Number of ghost cells
Returns
-------
int
ghost cell count
"""
return self._n_ghosts
@n_ghosts.setter
def n_ghosts(self, val: int) -> None:
self._n_ghosts = val
@property
def size(self) -> int:
"""Placeholder size
Returns
-------
int
length of arrays in placeholder
"""
return self._size
@size.setter
def size(self, val: int) -> None:
self._size = val
if self._size == 0:
self.__reset_sizes()
@property
def variables(self) -> dict[Any, Any]:
"""Defined container variables
Returns
-------
dict
list of variables and associated data
"""
return self._variables
def _define_variables(self):
self._variables = {}
[docs]
def setup(self, **kwargs: Any) -> None:
"""Sets up container according to specifications
Parameters
----------
**kwargs: Any :
keyword-value pair data, see :func:'set_from_key_value'
"""
if len(kwargs) > 0:
self.__reset_sizes()
self.set(**kwargs)
if self.__backup_data:
self.backup()
if self._last <= self._first:
self._last = self._first + self._size
self._allocate_placeholders()
if self.__backup_data:
self.get_from_backup()
self._pointers = []
self._keys = []
for var in self._variables.keys():
self._pointers.append(self._variables[var]['ref'])
self._keys.append(var)
self._n_var = len(self._pointers)
def _allocate_placeholders(self) -> None:
pass
[docs]
def de_allocate_placeholders(self) -> None:
"""De allocates placeholders - nullifies all arrays"""
self.__reset_sizes()
self.setup()
[docs]
def is_allocated(self) -> bool:
"""Placeholder allocation status
Returns
-------
bool
allocation status
"""
return self._size > 0
[docs]
def get_range(self) -> Tuple[int, int]:
"""Gets placeholder range (first, last)
Returns
-------
tuple
placeholder range (first, last)
"""
return self._first, self._last
[docs]
def GetNGhosts(self) -> int:
"""Gets number of allocated ghost cells
Returns
-------
int
number of ghost cells
"""
return self._size - (self._last - self._first)
[docs]
def write_to_file(self, f: BinaryIO) -> None:
"""Writes placeholder variable arrays to file (binary format)
Parameters
----------
f: BinaryIO :
file handle (binary file)
"""
for var in self._variables.keys():
f.write(bytes(self._variables[var]['ref'][self._first:self._last]))
[docs]
def read_from_file(self, f: BinaryIO) -> None:
"""Reads placeholder variable arrays from file (binary format)
Parameters
----------
f: BinaryIO :
file handle (binary file)
"""
N = self._last - self._first
for var in self._variables.keys():
self._variables[var]['ref'][self._first:self._last] = np.array(struct.unpack('d' * N, f.read(N * 8)))
[docs]
def print_flow_data(self, idx: int) -> None:
"""Prints stored variable information for specified cell index to stdout
Parameters
----------
idx: int :
cell index
"""
for var in self._variables.keys():
print(' %10s: %10.5E [%s]' % (self._variables[var]['longname'],
self._variables[var]['ref'][idx],
self._variables[var]['dimension']))
[docs]
def nullify(self) -> None:
"""Nullifies all array elements for all defined variables"""
for var in self._variables.keys():
self._variables[var]['ref'][:] = 0.
[docs]
def nullify_by_index(self, idx: int) -> None:
"""Nullifies all variables for specified cell index
Parameters
----------
idx: int :
cell index
"""
for var in self._variables.keys():
self._variables[var]['ref'][idx] = 0.
def __nullify_extreme_values(self) -> None:
for var in self._variables.keys():
self._variables[var]['min_val'] = 1.e20
self._variables[var]['max_val'] = -1.e20
[docs]
def scale(self, idx: int, sf: float) -> None:
"""Scales all variable values for specified cell index
Parameters
----------
idx: int :
cell index
sf: float :
scale factor
"""
for var in self._variables.keys():
self._variables[var]['ref'][idx] = self._variables[var]['ref'][idx] * sf
[docs]
def vector_scale(self, v: ndarray) -> None:
"""Scales all variable values for all elements using specified scaling array
Parameters
----------
v: ndarray :
scale factors for all cells
"""
for var in self._variables.keys():
self._variables[var]['ref'][self._first:self._last] = (self._variables[var]['ref'][self._first:self._last]
* v[self._first:self._last])
[docs]
def get_pointers(self) -> list[ndarray]:
"""Gets list of array pointers
Returns
-------
list[ndarray]
pointers to flow data arrays
"""
return self._pointers
[docs]
def get_keys(self) -> list:
"""Gets defined variable keys
Returns
-------
list
defined variable keys
"""
return self._keys
[docs]
def get_array(self, var: str = 'var') -> ndarray:
"""Get array for specified variable string
Parameters
----------
var : str :
variable string (Default value = 'var')
"""
return self._variables[var]['ref'][self._first:self._last]
[docs]
def set_from_array(self, from_array: ndarray, var: str = 'var'):
"""Set flow data for specified variable from array
Parameters
----------
from_array: ndarray :
prototype data array
var: str :
variable name (Default value = 'var')
"""
self._variables[var]['ref'][self._first:self._last] = from_array[self._first:self._last]
[docs]
def get_array_with_ghosts(self, var: str = 'var') -> ndarray:
"""Get array for specified variable string including ghost cell values
Parameters
----------
var : str :
variable string (Default value = 'var')
"""
return self._variables[var]['ref']
[docs]
def get_value(self, idx: int, var: str = 'var') -> float:
"""Get variable value for specified cell index
Parameters
----------
idx: int :
cell index
var: str :
variable name (Default value = 'var')
"""
return self._variables[var]['ref'][idx]
[docs]
def set_value(self, idx: int, val: float, var: str = 'var') -> None:
"""Sets variable value for specified cell index
Parameters
----------
idx: int :
cell index
val: float :
variable value
var: str :
variable name (Default value = 'var')
"""
self._variables[var]['ref'][idx] = val
[docs]
def initialize(self, val: float = 0., var: str = 'all') -> None:
"""Initialize placeholder
Parameters
----------
val: float :
variable value (Default value = 0.)
var: str :
variable name (Default value = 'all')
"""
if var == 'all':
for tmp in self._variables.keys():
self._variables[tmp]['ref'][self._first:self._last] = val
else:
self._variables[var]['ref'][self._first:self._last] = val
[docs]
def copy(self, fd: 'FlowDataPlaceHolder') -> None:
"""Copies data from prototype FlowDataPlaceHolder object
Parameters
----------
fd: FlowDataPlaceHolder :
prototype FlowDataPlaceHolder object
"""
self.setup(n_cells=fd.size)
tmp = fd.variables
for var in self._variables.keys():
self._variables[var]['ref'][self._first:self._last] = tmp[var]['ref'][self._first:self._last]
[docs]
def copy_values(self, from_idx: int, to_idx: int) -> None:
"""Copies from one array element to another
Parameters
----------
from_idx: int :
data source cell index
to_idx: int :
data destination cell index
"""
for var in self._variables.keys():
self._variables[var]['ref'][to_idx] = self._variables[var]['ref'][from_idx]
def __get_extreme_values(self, var: str) -> None:
self._variables[var]['min_val'] = np.min(self._variables[var]['ref'][self._first:self._last])
self._variables[var]['max_val'] = np.max(self._variables[var]['ref'][self._first:self._last])
def __get_max_values(self, var: str) -> None:
self._variables[var]['max_val'] = np.max(self._variables[var]['ref'][self._first:self._last])
def __get_min_values(self, var: str) -> None:
self._variables[var]['min_val'] = np.min(self._variables[var]['ref'][self._first:self._last])
def __get_max_norm(self, var: str) -> None:
self._variables[var]['max_norm'] = np.max(np.abs(self._variables[var]['ref'][self._first:self._last]))
def __print_extreme_values(self, var: str) -> None:
print(' %4s (min) [max]: (% 10.5E) [% 10.5E]' % (var,
self._variables[var]['min_val'],
self._variables[var]['max_val']))
def __print_extreme_values_by_type(self, var: str, value_type: str) -> None:
print(' %4s [%s]: % 10.5E' % (var, value_type, self._variables[var][value_type]))
[docs]
def get_extreme_values(self, var: str = 'all') -> None:
"""Updates placeholder variables extreme values (min/max)
Parameters
----------
var: str :
variable name (Default value = 'all')
"""
if type(var) is str:
if var == 'all':
for tmp in self._variables.keys():
self.__get_extreme_values(tmp)
else:
self.__get_extreme_values(var)
elif type(var) is list:
for tmp in var:
self.__get_extreme_values(tmp)
[docs]
def get_min_values(self, var: str = 'all') -> None:
"""Updates placeholder min values
Parameters
----------
var: str :
variable name (Default value = 'all')
"""
if type(var) is str:
if var == 'all':
for tmp in self._variables.keys():
self.__get_min_values(tmp)
else:
self.__get_min_values(var)
elif type(var) is list:
for tmp in var:
self.__get_min_values(tmp)
[docs]
def get_max_values(self, var: str = 'all') -> None:
"""Updates placeholder max values
Parameters
----------
var: str :
variable name (Default value = 'all')
"""
if type(var) is str:
if var == 'all':
for tmp in self._variables.keys():
self.__get_max_values(tmp)
else:
self.__get_max_values(var)
elif type(var) is list:
for tmp in var:
self.__get_max_values(tmp)
[docs]
def get_max_norm(self, var: str = 'all') -> None:
"""Updates variable max norm values (max absolute values)
Parameters
----------
var: str :
variable name (Default value = 'all')
"""
if type(var) is str:
if var == 'all':
for tmp in self._variables.keys():
self.__get_max_norm(tmp)
else:
self.__get_max_norm(var)
elif type(var) is list:
for tmp in var:
self.__get_max_norm(tmp)
[docs]
def normalize_extreme_values(self, norm_values: Union[list[float], ndarray], value_type: str = 'max_val') -> None:
"""Normalizes extreme values
Parameters
----------
norm_values: list[float] or ndarray :
normalization values (variable specific values)
value_type: str :
type of extreme value (Default value = 'max_val')
"""
ivar = 0
for tmp in self._variables.keys():
self._variables[tmp][value_type] /= norm_values[ivar]
ivar += 1
[docs]
def get_extreme_value_array(self, value_type: str = 'max_val') -> ndarray:
"""Gets array with calculated variable extreme values
Parameters
----------
value_type: str :
type of extreme values (Default value = 'max_val')
"""
return np.array([self._variables[var][value_type] for var in self._variables.keys()])
[docs]
def print_extreme_values_by_type(self, value_type: str = 'max_val', var: str = 'all') -> None:
"""Prints calculated extreme values by type to stdout
Parameters
----------
value_type: str :
extreme value type (Default value = 'max_val')
var: str :
variable name (Default value = 'all')
"""
if type(var) is str:
if var == 'all':
for tmp in self._variables.keys():
self.__print_extreme_values_by_type(tmp, value_type)
else:
self.__print_extreme_values_by_type(var, value_type)
elif type(var) is list:
for tmp in var:
self.__print_extreme_values_by_type(tmp, value_type)
[docs]
def print_extreme_values(self, var: Union[str, list[str]] = 'all') -> None:
"""Prints calculated extreme values to stdout
Parameters
----------
var: str or list[str] :
variable name(s) (Default value = 'all')
"""
if type(var) is str:
if var == 'all':
for tmp in self._variables.keys():
self.__get_extreme_values(tmp)
self.__print_extreme_values(tmp)
else:
self.__get_extreme_values(var)
self.__print_extreme_values(var)
elif type(var) is list:
for tmp in var:
self.__get_extreme_values(tmp)
self.__print_extreme_values(tmp)
[docs]
def print_extreme_value_locations(self) -> None:
"""Prints extreme values including location to stdout"""
for var in self._variables.keys():
print(' %4s (min loc) [max loc]: (%d) [%d]' %
(var, np.argmin(self._variables[var]['ref'][self._first:self._last]),
np.argmax(self._variables[var]['ref'][self._first:self._last])))
[docs]
def export(self, name: str, var: str = 'all') -> dict[Any, Any]:
"""Exports placeholder data to file (binary)
Parameters
----------
name: str :
file name base
var: str :
variable name (Default value = 'all')
"""
if type(var) is str:
if var == 'all':
for tmp in self._variables.keys():
self.__export_var(name, tmp)
else:
self.__export_var(name, var)
elif type(var) is list:
for tmp in var:
self.__export_var(name, tmp)
return self._variables
def __export_var(self, name: str, var: str) -> None:
g3d_disp = G3DDisplay()
self._variables[var]['filename'] = name + '_' + var + '.flow'
if g3d_disp.verbose:
print('exporting flow data to file: %s' % self._variables[var]['filename'])
with open(self._variables[var]['filename'], 'w+b') as f:
f.write(self._variables[var]['ref'][self._first:self._last].tobytes())
[docs]
def import_data(self, name: str, var: str = 'all') -> dict[Any, Any]:
"""Imports placeholder data from file (binary)
Parameters
----------
name: str :
file name base
var: str :
variable name (Default value = 'all')
"""
if type(var) is str:
if var == 'all':
for tmp in self._variables.keys():
self.__import_var_data(name, tmp)
else:
self.__import_var_data(name, var)
elif type(var) is list:
for tmp in var:
self.__import_var_data(name, tmp)
return self._variables
def __import_var_data(self, name: str, var: str) -> None:
g3d_disp = G3DDisplay()
self._variables[var]['filename'] = name + '_' + var + '.flow'
if g3d_disp.verbose:
print('importing flow data from file: %s' % self._variables[var]['filename'])
with open(self._variables[var]['filename'], 'r+b') as f:
self._variables[var]['ref'][self._first:self._last] = np.fromfile(f, dtype=float, count=self._size)
[docs]
def import_var_data_from_binary(self, f: BinaryIO, var: Union[list[str], str] = 'all') -> dict[Any, Any]:
"""Imports variable data from binary file
Parameters
----------
f: BinaryIO :
file handle
var: list[str] or str :
variable name(s) (Default value = 'all')
Returns
-------
dict
variables dict
"""
# imports data from provided file handle
if type(var) is str:
if var == 'all':
self.__import_var_data_from_binary(f, self._keys)
else:
self.__import_var_data_from_binary(f, var)
else:
self.__import_var_data_from_binary(f, var)
return self._variables
def __import_var_data_from_binary(self, f: BinaryIO, variables: Union[list[str], str]) -> None:
# imports data from provided file handle
if type(variables) is list:
for i in range(self._n_cells):
for var in variables:
self._variables[var]['ref'][i] = np.fromfile(f, dtype=float, count=1)
elif type(variables) is str:
self._variables[variables]['ref'][self._first:self._last] = np.fromfile(f, dtype=float, count=self._n_cells)
[docs]
def WAXPY(self, A: float, X: 'FlowDataPlaceHolder', Y: 'FlowDataPlaceHolder') -> None:
"""Vector operation W = a * X + Y where X and Y are instances of FlowDataPlaceholder and W is this placeholder
Parameters
----------
A: float :
scaling factor for vector X (scalar value)
X: FlowDataPlaceHolder :
flow data container
Y: FlowDataPlaceHolder :
flow data container
"""
X_pointers = X.get_pointers()
Y_pointers = Y.get_pointers()
for i in range(self._n_var):
self._pointers[i][self._first:self._last] = (
A * X_pointers[i][self._first:self._last] + Y_pointers[i][self._first:self._last])
[docs]
def AXPBY(self, A: float, B: float, Y: 'FlowDataPlaceHolder') -> None:
"""Vector operation a * X + b * Y where Y is an instance of FlowDataPlaceholder and X is this placeholder
Parameters
----------
A: float :
scaling factor for vector X (scalar value)
B: float :
scaling factor for vector Y (scalar value)
Y: FlowDataPlaceHolder :
flow data container
"""
Y_pointers = Y.get_pointers()
for i in range(self._n_var):
self._pointers[i][self._first:self._last] = (
A * self._pointers[i][self._first:self._last] + B * Y_pointers[i][self._first:self._last])
[docs]
def AXPY(self, A: float, Y: 'FlowDataPlaceHolder') -> None:
"""Vector operation a * X + Y where Y is an instance of FlowDataPlaceholder and X is this placeholder
Parameters
----------
A: float :
scaling factor for vector X (scalar value)
Y: FlowDataPlaceHolder :
flow data container
"""
Y_pointers = Y.get_pointers()
for i in range(self._n_var):
self._pointers[i][self._first:self._last] = (
A * self._pointers[i][self._first:self._last] + Y_pointers[i][self._first:self._last])
[docs]
def point_wise_scale(self, sf: ndarray) -> None:
"""Element data scaling using provided scaling factor array
Parameters
----------
sf: ndarray :
scaling factors
"""
for i in range(self._n_var):
self._pointers[i][self._first:self._last] = (
sf[self._first:self._last] * self._pointers[i][self._first:self._last])