"""CFlow variable placeholder"""
from typing import Union, Any
import os
if os.environ.get('__USE_CUDA', '0') == '1':
import cupy as np
else:
import numpy as np
from g3d.defs import G3DEntity
from g3d.defs import CFlowVariableType
[docs]
class CFlowVar(G3DEntity):
"""CFlow variable placeholder"""
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""initialize CFlow variable placeholder"""
self._type = CFlowVariableType.UNDEFINED
self._dim = ""
self._name = "VAR"
self._long_name = "CFlow variable"
self._default = 1.
self.__bak = 1.
self._value = self._default
self._tmp = None
super().__init__(*args, **kwargs)
def _set_default(self) -> None:
super()._set_default()
self._tag = 'cflow_variable'
self.nullify()
[docs]
def nullify(self) -> None:
"""nullifies CFlow variable placeholder"""
self._value = self._default
self._tmp = None
def __update_dim(self) -> None:
dim_array = ["", "m/s", "K", "Pa", "kg/m^3", "kg/s", "m^2", "m", "m", "rad", "deg"]
self._dim = dim_array[self._type.value - 1]
[docs]
def print(self) -> None:
"""prints CFlow variable data"""
print("%s: %s %.4e [%s]" % (self._name, self._long_name, self._value, self._dim))
[docs]
def init(self) -> None:
"""initialize CFlow variable placeholder temporary storage"""
self._tmp = self._value
[docs]
def backup(self) -> None:
"""backup CFlow variable placeholder data"""
self.__bak = self._value
[docs]
def restore(self) -> None:
"""get CFlow variable placeholder data from backup"""
self._value = self.__bak
@property
def type(self) -> Any:
"""CFlow variable placeholder type"""
return self._type
@type.setter
def type(self, val: Any) -> None:
if type(val) is CFlowVariableType:
self._type = val
self.__update_dim()
@property
def value(self) -> float:
"""CFlow variable placeholder value"""
return self._value
@value.setter
def value(self, val: float) -> None:
self._value = val
@property
def tmp(self) -> float:
"""CFlow variable placeholder temporary data"""
return self._tmp
@tmp.setter
def tmp(self, val: float) -> None:
self._tmp = val
@property
def dim(self) -> str:
"""CFlow variable placeholder dimension"""
return self._dim
@dim.setter
def dim(self, val: str) -> None:
self._dim = val
@property
def default(self) -> float:
"""CFlow variable placeholder default value"""
return self._default
@default.setter
def default(self, val: float) -> None:
self._default = val
@property
def name(self) -> str:
"""CFlow variable placeholder name"""
return self._name
@name.setter
def name(self, name: str) -> None:
self._name = name
@property
def long_name(self) -> str:
"""CFlow variable placeholder long name"""
return self._long_name
@long_name.setter
def long_name(self, name: str) -> None:
self._long_name = name
[docs]
def set_from_key_value(self, key: str, value: Any) -> bool:
"""CFlow variable placeholder set data from key => value input"""
if super().set_from_key_value(key, value):
return True
elif key == "type":
self._type = value
self.__update_dim()
return True
elif key == "value":
self._value = value
return True
elif key == "default":
self._default = value
self._value = value
return True
elif key == "name":
self._name = value
return True
elif key == "long_name":
self._long_name = value
return True
elif key == "dim":
self._dim = value
return True
return False
[docs]
def get_from_array(self, idx: int) -> None:
"""CFlow variable placeholder get value from temporary storage"""
self._value = self._tmp[idx]
[docs]
class CFlowVariables(G3DEntity):
"""CFlow variable set placeholder"""
def __init__(self, *args, **kwargs):
"""initialize CFlow variable set placeholder"""
self._vars = []
self._nvars = 0
self._nfvars = 0
self._isscalar = False
self._size = 0
super().__init__(*args, **kwargs)
@property
def nvars(self) -> int:
"""
number of variables in CFlow variable set
:return: number of variables (int)
"""
return self._nvars
@property
def nfvars(self) -> int:
"""
number of float variables
:return: number of float variables (int)
"""
return self._nfvars
@property
def size(self) -> int:
"""
array length of variables in CFlow variable set (if applicable)
:return: size (int)
"""
return self._size
@property
def isscalar(self) -> bool:
"""
scalar/vector status for current CFlow variable set
:return: scalar status (bool)
"""
return self._isscalar
def _set_default(self) -> None:
super()._set_default()
self._tag = 'cflow_variable_placeholder'
self.__nullify()
[docs]
def init(self, names: list[str]) -> None:
"""initialize placeholder vectors"""
self.__nullify()
for name in names:
for var in self._vars:
if var.name == name and var.tmp is None:
var.init()
self._fvars.append(var)
self._nfvars = len(self._fvars)
def __nullify(self) -> None:
self.nullify()
self._fvars = []
self._nfvars = 0
self._size = 1
self._isscalar = False
[docs]
def nullify(self, name: Union[str, int] = "all") -> None:
"""
nullify specified variables
input: name (default value = all => all variables will be nullified)
example: nullify(name)
"""
if type(name) in [int, np.int64, np.int32]:
if name < self._nvars:
self._vars[name].nullify()
else:
for var in self._vars:
if var.name == name or name == "all":
var.nullify()
[docs]
def get_from_args(self, *args: Any) -> None:
"""
get variable values from argument list
input: args
example: get_from_args(args)
"""
if len(args) > 0:
N = min([self._nfvars, len(args)])
for i in range(N):
value = args[i]
if type(value) is np.ndarray:
if value.size == 1:
value = np.float64(value)
self._fvars[i].tmp = value
[docs]
def arrayify(self) -> None:
"""
make all variables arrays if one or more variables are specified as arrays
"""
self._isscalar = True
self._size = 1
for var in self._fvars:
if not np.isscalar(var.tmp):
self._isscalar = False
self._size = max([self._size, len(var.tmp)])
if self._isscalar:
for var in self._fvars:
var.tmp = np.asarray([var.tmp])
else:
for var in self._fvars:
if np.isscalar(var.tmp):
var.tmp = np.ones(self._size) * var.tmp
else:
var.tmp = np.asarray(var.tmp)
[docs]
def backup(self) -> None:
"""
backup all float variables
"""
for var in self._fvars:
var.backup()
[docs]
def restore(self) -> None:
"""
restore all float variables from backup
"""
for var in self._fvars:
var.restore()
[docs]
def get_from_array(self, idx: int) -> None:
"""get value from array"""
for var in self._fvars:
var.get_from_array(idx)
[docs]
def set_from_key_value(self, key: str, value: Any) -> bool:
"""
set CFlow variable data from key => value input
"""
if super().set_from_key_value(key, value):
return True
else:
for var in self._vars:
if var.name == key:
if type(value) is np.ndarray:
if value.size == 1:
value = np.float64(value)
var.value = value
if var.tmp is not None:
var.tmp = var.value
return True
return False
[docs]
def add(self, *args: Any, **kwargs: Any) -> CFlowVar:
"""
add variable to variable set
example: add(name="T", long_name="Temperature", type=CFlowVariableType.TEMPERATURE)
"""
var = CFlowVar(*args, **kwargs)
self._vars.append(var)
self._nvars = len(self._vars)
return var
[docs]
def get(self, name: Union[str, int]) -> Union[CFlowVar, None]:
"""
get variable with specified name
input: name
example: var = get(name)
"""
if type(name) in [int, np.int64, np.int32]:
if name < self._nvars:
return self._vars[name]
else:
for var in self._vars:
if var.name == name:
return var
return None
[docs]
def get_val(self, name: Union[str, int]) -> Union[float, None]:
"""
get value for variable with specified name
input: name
example: val = get_val(name)
"""
if type(name) in [int, np.int64, np.int32]:
if name < self._nvars:
return self._vars[name].value
else:
for var in self._vars:
if var.name == name:
return var.value
return None
[docs]
def set_val(self, name: Union[str, int], value: float) -> None:
"""
set value for variable with specified name
input: name, value
example: set_val(name, value)
"""
if type(name) in [int, np.int64, np.int32]:
if name < self._nvars:
self._vars[name].value = value
else:
for var in self._vars:
if var.name == name:
var.value = value
[docs]
def set_default(self, name: Union[str, int], value: float) -> None:
"""
set default value for variable with specified name
input: name, value
example: set_default(name, value)
"""
if type(name) in [int, np.int64, np.int32]:
if name < self._nvars:
self._vars[name].default = value
else:
for var in self._vars:
if var.name == name:
var.default = value
[docs]
def print(self, name: Union[str, int] = "all") -> None:
"""
print variable information for all or specified variables in variable set
input: name (default value="all")
example: print(name)
"""
if type(name) in [int, np.int64, np.int32]:
if name < self._nvars:
self._vars[name].print()
else:
for var in self._vars:
if var.name == name or name == "all":
var.print()
[docs]
class CFlowStack(G3DEntity):
"""CFlow variables stack"""
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""initialize CFlow variables stack"""
self._vars = CFlowVariables()
super().__init__(*args, **kwargs)
def _set_default(self) -> None:
super()._set_default()
self._tag = 'g3d_stack'
@property
def vars(self) -> CFlowVariables:
"""CFlow variables placeholder"""
return self._vars
[docs]
def nullify(self) -> None:
"""nullify CFlow variables placeholder"""
self._vars.nullify()
[docs]
def init(self, names: list[str]) -> None:
"""initialize CFlow variables placeholder with specified names"""
self._vars.init(names)
[docs]
def print(self) -> None:
"""nullify CFlow variables placeholder data"""
self._vars.print()