"""solver environment (SolverEnv class definition)"""
from typing import Any, Union, Tuple
import os
if os.environ.get('__USE_CUDA', '0') == '1':
from cupy import ndarray
else:
from numpy import ndarray
from g3d.defs import G3DWorld
from g3d.defs import G3DCode
from g3d.defs import CodeType
from g3d.defs import ShockDampingMode
from g3d.defs import TVDLimiters
from g3d.calc import EmbeddedCFlow
from g3d.flow.placeholders import MatrixRows, FlowDataPlaceHolder
from g3d.flow.domain import Domain
from g3d.flow.solver import Solver
from g3d.flow.io import G3DIO
from g3d.gas import Gas
from g3d.flow.vars import Cons
from g3d.flow.vars import Auxv
from g3d.flow.geo import Q1DPipe
from g3d.mesh.geo import IJKRegion
from g3d.flow.domain.boco.boundaries import BoCo
from g3d.flow.domain.boco.bc_mod import BCMod
[docs]
class SolverEnv(G3DCode):
def __init__(self, *args: Any, **kwargs: Any) -> None:
super().__init__(*args, **kwargs)
def _set_code_type(self) -> None:
g3d_world = G3DWorld()
g3d_world.info.code_type = CodeType.G3D_SOLVER
g3d_world.info.code_name = 'g3dflow.py'
g3d_world.info.code_description = 'a python-based solver for compressible flows'
g3d_world.options.short_opt += ''
g3d_world.options.long_opt.append('nstep=')
g3d_world.options.long_opt.append('restart')
g3d_world.options.long_opt.append('no_restart')
g3d_world.options.long_opt.append('print_iterations')
def _set_default(self) -> None:
super()._set_default()
self._tag = 'g3d_solver'
self._calculator = EmbeddedCFlow()
self.__dom = Domain()
self.__solver = Solver()
self.__io = G3DIO()
self.__first_run = True
self._calculator.gas = self.__dom.gas
def _print_help_message(self) -> None:
super()._print_help_message()
def _evaluate_run_option(self, opt: str, arg: Any) -> None:
g3d_world = G3DWorld()
super()._evaluate_run_option(opt, arg)
if opt in ['--nstep']:
g3d_world.options.nstep = arg
elif opt in ['--restart']:
g3d_world.options.restart = True
elif opt in ['--no_restart']:
g3d_world.options.restart = False
elif opt in ['--print_iterations']:
g3d_world.display.print_iterations = True
[docs]
def set_from_key_value(self, key: str, value: Any) -> bool:
g3d_world = G3DWorld()
if super().set_from_key_value(key, value):
if key == 'jobname' or key == 'wd' or key == 'working_directory':
self.__solver.reset_solver_status()
return True
elif key == 'progress_bar':
g3d_world.display.progress_bar = value
return True
elif key == 'print_iterations':
g3d_world.display.print_iterations = value
return True
elif key == 'plot_residuals':
g3d_world.options.auto_plot_residuals = value
return True
elif key == 'restart':
g3d_world.options.restart = value
return True
elif key == 'nstep':
g3d_world.options.nstep = value
return True
elif key == 'nstep_save':
g3d_world.options.nstep_save = value
return True
elif key == 'save_all_files':
g3d_world.io.save_all_files = value
return True
elif key == 'eqn':
g3d_world.options.eqn_mode = value
self.__solver.change_eqn_mode()
return True
elif key == 'cfl':
g3d_world.num.time.cfl = value
g3d_world.options.use_irsm = g3d_world.num.time.cfl > g3d_world.num.irsm.cfl_crit
return True
elif key == 'dtime':
g3d_world.num.time.dtime = value
return True
elif key == 'time_mode':
g3d_world.num.time.time_mode = value
return True
elif key == 'n_sub_iter':
g3d_world.num.time.n_sub_iter = value
return True
elif key == 'time_stepper_mode':
g3d_world.num.time.time_stepper_mode = value
return True
elif key == 'shock_damp_coef' or key == 'damp_p':
g3d_world.num.scheme.shock_damp_coef = value
return True
elif key == 'entropy_damp_coef' or key == 'damp_c':
g3d_world.num.scheme.entropy_damp_coef = value
return True
elif key == 'convective_scheme':
g3d_world.num.scheme.convective_scheme = value
return True
elif key == 'tvd':
if value is TVDLimiters.TVD_RO:
g3d_world.num.scheme.use_density_tvd_limiter = True
if value is TVDLimiters.TVD_TURB:
g3d_world.num.scheme.use_turbulence_model_tvd_limiter = True
elif key == 'use_density_TVD':
g3d_world.num.scheme.use_density_tvd_limiter = value
return True
elif key == 'use_turb_TVD':
g3d_world.num.scheme.use_turbulence_model_tvd_limiter = value
return True
elif key == 'break_residual':
g3d_world.options.break_residual = value
return True
elif key == 'use_velocity_gradients':
g3d_world.options.use_velocity_gradients = value
return True
elif key == 'use_irsm':
g3d_world.options.use_irsm = value
return True
elif key == 'irsm_lambda':
g3d_world.num.irsm.irsm_lambda = value
return True
elif key == 'shock_damping_mode':
g3d_world.num.shock_damping_mode = value
if value is ShockDampingMode.DUCROSS_CELL_GRADIENTS:
g3d_world.options.use_velocity_gradients = True
return True
elif key == 'q3d_source_factor':
g3d_world.options.q3d_source_factor = value
return True
elif key == 'turbulence_model':
g3d_world.models.turbulence_model = value
return True
elif key == 'p_min':
g3d_world.num.limiters.p_min = value
return True
elif key == 'p_max':
g3d_world.num.limiters.p_max = value
return True
return False
@property
def calculator(self) -> EmbeddedCFlow:
"""
:return:
"""
return self._calculator
@property
def solver(self) -> Solver:
"""
:return:
"""
return self.__solver
@property
def dom(self) -> Domain:
"""
:return:
"""
return self.__dom
[docs]
def jobname_append(self, name: str) -> None:
self.__solver.reset_solver_status()
super().jobname_append(name)
[docs]
@staticmethod
def get_new_Cons() -> Cons:
"""
:return:
"""
q = Cons()
return q
[docs]
@staticmethod
def get_new_Auxv() -> Auxv:
"""
:return:
"""
p = Auxv()
return p
[docs]
@staticmethod
def get_new_BoCo(*args: Any, **kwargs: Any) -> BoCo:
"""
:param args:
:param kwargs:
:return:
"""
a = BoCo(*args, **kwargs)
return a
[docs]
def add_BCMod(self, *args: Any, **kwargs: Any) -> BCMod:
"""
:param args:
:param kwargs:
:return:
"""
r = BCMod(*args, **kwargs)
self.__dom.bc_mods.append(r)
return r
[docs]
@staticmethod
def get_new_Q1DPipe() -> Q1DPipe:
"""
:return:
"""
q = Q1DPipe()
return q
[docs]
@staticmethod
def get_new_Region() -> IJKRegion:
"""
:return:
"""
r = IJKRegion()
return r
[docs]
@staticmethod
def get_new_matrix() -> MatrixRows:
"""
:return:
"""
m = MatrixRows()
return m
[docs]
def auxvr(self) -> None:
"""
:return:
"""
return self.__dom.auxvr()
[docs]
def get_cons_arrays(self) -> Tuple[ndarray, ...]:
"""
:return:
"""
return self.__dom.get_cons_arrays()
[docs]
def get_cons_array(self, var: str = 'ro') -> ndarray:
"""
:param var:
:return:
"""
return self.__dom.get_cons_array(var)
[docs]
def get_auxv_arrays(self) -> Tuple[ndarray, ...]:
"""
:return:
"""
return self.__dom.get_auxv_arrays()
[docs]
def get_auxv_array(self, var: str = 'p') -> ndarray:
"""
:param var:
:return:
"""
return self.__dom.get_auxv_array(var)
[docs]
def get_post_arrays(self) -> Tuple[ndarray, ...]:
"""
:return:
"""
return self.__dom.get_post_arrays()
[docs]
def get_post_array(self, var: str = 'M') -> ndarray:
"""
:param var:
:return:
"""
return self.__dom.get_post_array(var)
[docs]
def get_node_coordinates(self) -> Tuple[ndarray, ...]:
"""
:return:
"""
return self.__dom.get_node_coordinates()
[docs]
def get_cell_centers(self) -> Tuple[ndarray, ...]:
"""
:return:
"""
return self.__dom.get_cell_centers()
[docs]
def get_volume(self, idx: int = None) -> Union[float, ndarray]:
"""
:param idx:
:return:
"""
return self.__dom.get_volume(idx)
[docs]
def get_streamtube_normal(self, idx: int = None) -> Union[float, ndarray]:
"""
:param idx:
:return:
"""
return self.__dom.get_streamtube_normal(idx)
[docs]
def get_block_sizes(self, im: int) -> Tuple[int, int, int]:
"""
:param im:
:return:
"""
return self.__dom.get_block_sizes(im)
[docs]
def get_node(self, im: int, i: int, j: int, k: int) -> int:
"""
:param im:
:param i:
:param j:
:param k:
:return:
"""
return self.__dom.get_node(im, i, j, k)
[docs]
def get_cell(self, im: int, i: int, j: int, k: int) -> int:
"""
:param im:
:param i:
:param j:
:param k:
:return:
"""
return self.__dom.get_cell(im, i, j, k)
[docs]
def get_shock_capturing_field(self) -> ndarray:
"""
:return:
"""
return self.__solver.get_shock_capturing_field()
[docs]
def run(self) -> None:
"""
"""
self.__setup_and_allocate()
self.__solver.run()
[docs]
def import_flow_data(self, *args: Any, **kwargs: Any) -> None:
"""
:param args:
:param kwargs:
"""
self.__setup_and_allocate()
self.__solver.import_flow_data(*args, **kwargs)
[docs]
def import_g3d_flow_data(self, *args: Any, **kwargs: Any) -> None:
"""
:param args:
:param kwargs:
"""
self.__setup_and_allocate()
self.__solver.import_g3d_flow_data(*args, **kwargs)
[docs]
def generate_mesh_from_pipe_data(self, q1d: Q1DPipe) -> None:
"""
:param q1d:
"""
self.__dom.generate_mesh_from_pipe_data(q1d)
[docs]
def export_mesh(self, *args: Any, **kwargs: Any) -> None:
"""
:param args:
:param kwargs:
:return:
"""
return self.__dom.export_mesh(*args, **kwargs)
[docs]
def import_mesh(self, *args: Any, **kwargs: Any) -> None:
"""
:param args:
:param kwargs:
"""
self.__dom.import_mesh(*args, **kwargs)
[docs]
def export_flow(self, *args: Any, **kwargs: Any) -> None:
"""
:param args:
:param kwargs:
"""
self.__setup_and_allocate()
self.__dom.export_flow(*args, **kwargs)
[docs]
def export_velocity_gradients(self, *args: Any, **kwargs: Any) -> None:
"""
:param args:
:param kwargs:
"""
self.__setup_and_allocate()
self.__dom.export_velocity_gradients(*args, **kwargs)
[docs]
def export_field(self, field: FlowDataPlaceHolder, **kwargs: Any) -> None:
"""
:param field:
:param kwargs:
"""
self.__setup_and_allocate()
self.__io.export_field(self.__dom, field, **kwargs)
[docs]
def export_scalar(self, var, **kwargs: Any) -> None:
"""
:param var:
:param kwargs:
"""
self.__setup_and_allocate()
self.__io.export_scalar(self.__dom, var, **kwargs)
[docs]
def export_scalars(self, variables, **kwargs: Any) -> None:
"""
:param variables:
:param kwargs:
"""
self.__setup_and_allocate()
self.__io.export_scalars(self.__dom, variables, **kwargs)
[docs]
def import_boundary_definitions(self, **kwargs: Any) -> None:
"""
:param kwargs:
:return:
"""
return self.__dom.import_boundary_definitions(**kwargs)
def __setup_and_allocate(self) -> None:
if self.__first_run:
self.__dom.setup_and_allocate()
self.__solver.setup_and_allocate(self.__dom)
self.__solver.associate(self.__dom)
self.__first_run = False
[docs]
def flux_check(self) -> None:
"""
"""
self.__setup_and_allocate()
g3d_world = G3DWorld()
g3d_world.display.hline()
print(' SOLVER FLUX CHECK')
g3d_world.display.hline()
self.__solver.flux_check()
[docs]
def diff_flux_check(self) -> None:
"""
"""
self.__setup_and_allocate()
g3d_world = G3DWorld()
g3d_world.display.hline()
print(' SOLVER FLUX CHECK')
g3d_world.display.hline()
self.__solver.diff_flux_check()
[docs]
def ltsp_test(self) -> None: # OBS! remove later
"""
"""
self.__setup_and_allocate()
g3d_world = G3DWorld()
g3d_world.display.hline()
print(' SOLVER LTSP TEST')
g3d_world.display.hline()
self.__solver.ltsp_test()
[docs]
def step_test(self) -> None: # OBS! remove later
"""
"""
self.__setup_and_allocate()
g3d_world = G3DWorld()
g3d_world.display.hline()
print(' SOLVER STEP TEST')
g3d_world.display.hline()
self.__solver.step_test()
[docs]
def check_cell_metrics(self, ic: int) -> None:
"""
:param ic:
"""
self.__setup_and_allocate()
g3d_world = G3DWorld()
g3d_world.display.hline()
print(' CHECK CELL METRICS (cell %d)' % ic)
self.__dom.check_cell_metrics(ic)
g3d_world.display.hline()
[docs]
def check_FMP_metrics(self) -> None:
"""
"""
self.__setup_and_allocate()
X, Y, Z = self.__dom.check_FMP_metrics()
self.__io.export_scalars(self.__dom, [X, Y, Z], variable_names=['SX', 'SY', 'SZ'], name='fmp_metrics_test')
[docs]
def export_inverse_cell_volume(self) -> None:
"""
"""
self.__setup_and_allocate()
voli = self.__dom.get_inverse_volume()
self.__io.export_scalar(self.__dom, voli, variable_name='inverse cell volume', name='inverse_cell_volume')
[docs]
def export_cell_volume(self) -> None:
"""
"""
self.__setup_and_allocate()
vol = self.__dom.get_volume()
self.__io.export_scalar(self.__dom, vol, variable_name='cell volume', name='cell_volume')
# Domain re-directs
[docs]
def cell_get_block(self, kc: int) -> int:
"""
:param kc:
:return:
"""
return self.__dom.cell_get_block(kc)
[docs]
def node_get_block(self, kp: int) -> int:
"""
:param kp:
:return:
"""
return self.__dom.node_get_block(kp)
[docs]
def cell_get_ijk(self, kc: int) -> Tuple[int, int, int]:
"""
:param kc:
:return:
"""
return self.__dom.cell_get_ijk(kc)
[docs]
def node_get_ijk(self, kp: int) -> Tuple[int, int, int]:
"""
:param kp:
:return:
"""
return self.__dom.node_get_ijk(kp)
[docs]
def update_BC(self, boco: BoCo) -> None:
"""
:param boco:
"""
self.__dom.update_BC(boco)
[docs]
def update_BC_family(self, *args: Any, **kwargs: Any) -> None:
"""
:param args:
:param kwargs:
"""
self.__dom.update_BC_family(*args, **kwargs)
[docs]
def update_BC_type(self, *args: Any, **kwargs: Any) -> None:
"""
:param args:
:param kwargs:
"""
self.__dom.update_BC_type(*args, **kwargs)
[docs]
def get_gas(self) -> Gas:
"""
:return:
"""
return self.__dom.gas
[docs]
def get_gas_properties(self) -> Tuple[float, float]:
"""
:return:
"""
return self.__dom.get_gas_properties()
[docs]
def set_gas_properties(self, g, R) -> None:
"""
:param g:
:param R:
"""
self.__dom.set_gas_properties(g, R)
[docs]
def initialize_flow_field(self, f: Union[Auxv, Cons]) -> None:
"""
:param f:
"""
if type(f) is Auxv:
self.__initialize_from_Auxv(f)
elif type(f) is Cons:
self.__initialize_from_Cons(f)
[docs]
def initialize_region_flow_field(self, r: IJKRegion, f: Union[Auxv, Cons]) -> None:
"""
:param r:
:param f:
"""
if type(f) is Auxv:
self.__initialize_region_from_Auxv(r, f)
elif type(f) is Cons:
self.__initialize_region_from_Cons(r, f)
[docs]
def initialize_block_flow_field(self, im: int, f: Union[Auxv, Cons]) -> None:
"""
:param im:
:param f:
"""
if type(f) is Auxv:
self.__initialize_block_from_Auxv(im, f)
elif type(f) is Cons:
self.__initialize_block_from_Cons(im, f)
def __initialize_from_Cons(self, q: Cons) -> None:
self.__dom.initsol_from_Cons(q)
def __initialize_from_Auxv(self, p: Auxv) -> None:
self.__dom.initsol_from_Auxv(p)
def __initialize_region_from_Cons(self, r: IJKRegion, q: Cons) -> None:
self.__dom.initsol_region_from_Cons(r, q)
def __initialize_region_from_Auxv(self, r: IJKRegion, p: Auxv) -> None:
self.__dom.initsol_region_from_Auxv(r, p)
def __initialize_block_from_Cons(self, im: int, q: Cons) -> None:
self.__dom.initsol_block_from_Cons(im, q)
def __initialize_block_from_Auxv(self, im: int, p: Auxv) -> None:
self.__dom.initsol_block_from_Auxv(im, p)
[docs]
def initialize_from_conservative(self, ro: float, ru: float, rv: float, rw: float, et: float) -> None:
"""
:param ro:
:param ru:
:param rv:
:param rw:
:param et:
"""
self.__dom.initsol_conservative(ro, ru, rv, rw, et)
[docs]
def initialize_from_primitive(self, p: float, u: float, v: float, w: float, t: float) -> None:
"""
:param p:
:param u:
:param v:
:param w:
:param t:
"""
self.__dom.initsol_primitive(p, u, v, w, t)
[docs]
def initialize_linear(self, ql: Cons, qr: Cons, direction: str = 'x') -> None:
"""
:param ql:
:param qr:
:param direction:
"""
self.__dom.initsol_linear(ql, qr, direction)
[docs]
def bc_update_speed_of_sound(self, boco: BoCo) -> None:
"""
:param boco:
"""
self.__dom.bc_update_speed_of_sound(boco)
[docs]
def bc_update_primitive(self, boco: BoCo) -> None:
"""
:param boco:
"""
self.__dom.bc_update_primitive(boco)
[docs]
def bc_update_conservative(self, boco: BoCo) -> None:
"""
:param boco:
"""
self.__dom.bc_update_conservative(boco)
# solver settings
[docs]
@staticmethod
def add_TVD_limiter(m: Any) -> None:
"""
:param m:
"""
if type(m) is TVDLimiters:
g3d_world = G3DWorld()
if m is TVDLimiters.TVD_RO:
g3d_world.num.scheme.use_density_tvd_limiter = True
if m is TVDLimiters.TVD_TURB:
g3d_world.num.scheme.use_turbulence_model_tvd_limiter = True
[docs]
@staticmethod
def remove_TVD_limiter(m: Any) -> None:
"""
:param m:
"""
if type(m) is TVDLimiters:
g3d_world = G3DWorld()
if m is TVDLimiters.TVD_RO:
g3d_world.num.scheme.use_density_tvd_limiter = False
if m is TVDLimiters.TVD_TURB:
g3d_world.num.scheme.use_turbulence_model_tvd_limiter = False
[docs]
@staticmethod
def add_to_nstep(N: int) -> None:
"""
:param N:
"""
g3d_world = G3DWorld()
g3d_world.options.nstep = g3d_world.options.nstep + N
[docs]
@staticmethod
def set_damp(fp: float, fc: float) -> None:
"""
:param fp:
:param fc:
"""
g3d_world = G3DWorld()
g3d_world.num.shock_damp_coef = fp
g3d_world.num.entropy_damp_coef = fc
[docs]
@staticmethod
def set_convection_scheme_coefs(c1: float, c2: float, c3: float, c4: float) -> None:
"""
:param c1:
:param c2:
:param c3:
:param c4:
"""
g3d_world = G3DWorld()
g3d_world.num.scheme.set_coefs(c1, c2, c3, c4)
[docs]
@staticmethod
def get_time() -> float:
"""
:return:
"""
g3d_world = G3DWorld()
return g3d_world.status.time
[docs]
@staticmethod
def get_residual() -> float:
"""
:return:
"""
g3d_world = G3DWorld()
return g3d_world.status.residual
[docs]
@staticmethod
def get_n_cells() -> int:
"""
:return:
"""
g3d_world = G3DWorld()
return g3d_world.sizes.n_cells
[docs]
@staticmethod
def get_n_nodes() -> int:
"""
:return:
"""
g3d_world = G3DWorld()
return g3d_world.sizes.n_nodes
[docs]
@staticmethod
def get_n_faces() -> int:
"""
:return:
"""
g3d_world = G3DWorld()
return g3d_world.sizes.n_faces
[docs]
@staticmethod
def get_n_ghost_cells() -> int:
"""
:return:
"""
g3d_world = G3DWorld()
return g3d_world.sizes.n_ghost_cells
[docs]
@staticmethod
def get_n_ghost_nodes() -> int:
"""
:return:
"""
g3d_world = G3DWorld()
return g3d_world.sizes.n_ghost_nodes
# def set_turbulence_model(self, m):
# g3d_world = G3DWorld()
# g3d_world.models.turbulence_model = m
# def get_turbulence_model(self):
# g3d_world = G3DWorld()
# return g3d_world.models.turbulence_model
[docs]
def plot_scalar_2D(self, var: ndarray, im: int = 0, k: int = 0, **kwargs: Any) -> None:
"""
:param var:
:param im:
:param k:
:param kwargs:
"""
self.__io.plot_scalar_2D(self.__dom, var, im, k, **kwargs)
[docs]
def plot_scalars_2D(self, variables: ndarray, im: int = 0, k: int = 0, **kwargs: Any) -> None:
"""
:param variables:
:param im:
:param k:
:param kwargs:
"""
self.__io.plot_scalars_2D(self.__dom, variables, im, k, **kwargs)
[docs]
def plot_field_2D(self, field: FlowDataPlaceHolder, im: int = 0, k: int = 0, **kwargs: Any) -> None:
"""
:param field:
:param im:
:param k:
:param kwargs:
"""
self.__io.plot_field_2D(self.__dom, field, im, k, **kwargs)
[docs]
def plot_flow_2D(self, im: int = 0, k: int = 0, **kwargs: Any) -> None:
"""
:param im:
:param k:
:param kwargs:
"""
self.__io.plot_flow_2D(self.__dom, self.__dom.cons,
self.__dom.auxv, self.__dom.post, im, k, **kwargs)
[docs]
def plot_probe_array(self, var: ndarray,
im: int, i0: int, j0: int, k0: int, idir: int, nci: int,
**kwargs: Any) -> None:
"""
:param var:
:param im:
:param i0:
:param j0:
:param k0:
:param idir:
:param nci:
:param kwargs:
"""
self.__io.plot_probe_array(self.__dom, var, im, i0, j0, k0, idir, nci, **kwargs)