"""
Flow360 simulation parameters
"""
from __future__ import annotations
from typing import Annotated, List, Literal, Optional, Union
import pydantic as pd
import unyt as u
from flow360.component.simulation.conversion import (
LIQUID_IMAGINARY_FREESTREAM_MACH,
unit_converter,
)
from flow360.component.simulation.framework.base_model import Flow360BaseModel
from flow360.component.simulation.framework.entity_registry import EntityRegistry
from flow360.component.simulation.framework.param_utils import (
AssetCache,
_set_boundary_full_name_with_zone_name,
_update_entity_full_name,
_update_zone_boundaries_with_metadata,
register_entity_list,
)
from flow360.component.simulation.framework.updater import updater
from flow360.component.simulation.framework.updater_utils import (
Flow360Version,
recursive_remove_key,
)
from flow360.component.simulation.meshing_param.params import MeshingParams
from flow360.component.simulation.meshing_param.volume_params import (
AutomatedFarfield,
RotationCylinder,
RotationVolume,
)
from flow360.component.simulation.models.surface_models import SurfaceModelTypes
from flow360.component.simulation.models.volume_models import (
ActuatorDisk,
BETDisk,
Fluid,
Solid,
VolumeModelTypes,
)
from flow360.component.simulation.operating_condition.operating_condition import (
OperatingConditionTypes,
)
from flow360.component.simulation.outputs.outputs import (
AeroAcousticOutput,
IsosurfaceOutput,
OutputTypes,
ProbeOutput,
SurfaceIntegralOutput,
SurfaceProbeOutput,
UserDefinedField,
VolumeOutput,
)
from flow360.component.simulation.primitives import (
ReferenceGeometry,
_SurfaceEntityBase,
_VolumeEntityBase,
)
from flow360.component.simulation.run_control.run_control import RunControl
from flow360.component.simulation.time_stepping.time_stepping import Steady, Unsteady
from flow360.component.simulation.unit_system import (
AbsoluteTemperatureType,
DensityType,
DimensionedTypes,
LengthType,
MassType,
TimeType,
UnitSystem,
UnitSystemType,
VelocityType,
is_flow360_unit,
unit_system_manager,
unyt_quantity,
)
from flow360.component.simulation.user_code.core.types import (
batch_get_user_variable_units,
get_post_processing_variables,
)
from flow360.component.simulation.user_defined_dynamics.user_defined_dynamics import (
UserDefinedDynamic,
)
from flow360.component.simulation.utils import model_attribute_unlock
from flow360.component.simulation.validation.validation_output import (
_check_output_fields,
_check_output_fields_valid_given_turbulence_model,
_check_unique_surface_volume_probe_entity_names,
_check_unique_surface_volume_probe_names,
_check_unsteadiness_to_use_aero_acoustics,
)
from flow360.component.simulation.validation.validation_simulation_params import (
_check_and_add_noninertial_reference_frame_flag,
_check_cht_solver_settings,
_check_complete_boundary_condition_and_unknown_surface,
_check_consistency_hybrid_model_volume_output,
_check_consistency_wall_function_and_surface_output,
_check_duplicate_entities_in_models,
_check_duplicate_isosurface_names,
_check_duplicate_surface_usage,
_check_hybrid_model_to_use_zonal_enforcement,
_check_low_mach_preconditioner_output,
_check_numerical_dissipation_factor_output,
_check_parent_volume_is_rotating,
_check_time_average_output,
_check_unsteadiness_to_use_hybrid_model,
_check_valid_models_for_liquid,
)
from flow360.error_messages import (
unit_system_inconsistent_msg,
use_unit_system_for_simulation_msg,
)
from flow360.exceptions import Flow360ConfigurationError, Flow360RuntimeError
from flow360.log import log
from flow360.version import __version__
from .validation.validation_context import (
CASE,
SURFACE_MESH,
VOLUME_MESH,
CaseField,
ConditionalField,
context_validator,
get_validation_info,
)
ModelTypes = Annotated[Union[VolumeModelTypes, SurfaceModelTypes], pd.Field(discriminator="type")]
class _ParamModelBase(Flow360BaseModel):
"""
Base class that abstracts out all Param type classes in Flow360.
"""
version: str = pd.Field(__version__, frozen=True)
unit_system: UnitSystemType = pd.Field(frozen=True, discriminator="name")
model_config = pd.ConfigDict(include_hash=True)
@classmethod
def _init_check_unit_system(cls, **kwargs):
"""
Check existence of unit system and raise an error if it is not set or inconsistent.
"""
if unit_system_manager.current is None:
raise Flow360RuntimeError(use_unit_system_for_simulation_msg)
# pylint: disable=duplicate-code
kwarg_unit_system = kwargs.pop("unit_system", None)
if kwarg_unit_system is not None:
if not isinstance(kwarg_unit_system, UnitSystem):
kwarg_unit_system = UnitSystem.from_dict(**kwarg_unit_system)
if kwarg_unit_system != unit_system_manager.current:
raise Flow360RuntimeError(
unit_system_inconsistent_msg(
kwarg_unit_system.system_repr(),
unit_system_manager.current.system_repr(),
)
)
return kwargs
@classmethod
def _get_version_from_dict(cls, model_dict: dict) -> str:
version = model_dict.get("version", None)
if version is None:
raise Flow360RuntimeError("Failed to find SimulationParams version from the input.")
return version
@classmethod
def _update_param_dict(cls, model_dict, version_to=__version__):
"""
1. Find the version from the input dict.
2. Update the input dict to `version_to` which by default is the current version.
3. If the simulation.json has higher version, then return the dict as is without modification.
Returns
-------
dict
The updated parameters dictionary.
bool
Whether the `model_dict` has higher version than `version_to` (AKA forward compatibility mode).
"""
input_version = cls._get_version_from_dict(model_dict=model_dict)
forward_compatibility_mode = Flow360Version(input_version) > Flow360Version(version_to)
if not forward_compatibility_mode:
model_dict = updater(
version_from=input_version,
version_to=version_to,
params_as_dict=model_dict,
)
return model_dict, forward_compatibility_mode
@staticmethod
def _sanitize_params_dict(model_dict):
"""
!!!WARNING!!!: This function changes the input dict in place!!!
Clean the redundant content in the params dict from WebUI
"""
recursive_remove_key(model_dict, "_id")
return model_dict
def _init_no_unit_context(self, filename, file_content, **kwargs):
"""
Initialize the simulation parameters without a unit context.
"""
if unit_system_manager.current is not None:
raise Flow360RuntimeError(
f"When loading params from file: {self.__class__.__name__}(filename), "
"unit context must not be used."
)
if filename is not None:
model_dict = self._handle_file(filename=filename, **kwargs)
else:
model_dict = self._handle_dict(**file_content)
model_dict = _ParamModelBase._sanitize_params_dict(model_dict)
# When treating files/file like contents the updater will always be run.
model_dict, _ = _ParamModelBase._update_param_dict(model_dict)
unit_system = model_dict.get("unit_system")
with UnitSystem.from_dict(**unit_system): # pylint: disable=not-context-manager
super().__init__(**model_dict)
def _init_with_unit_context(self, **kwargs):
"""
Initializes the simulation parameters with the given unit context.
"""
# When treating dicts the updater is skipped.
kwargs = _ParamModelBase._init_check_unit_system(**kwargs)
super().__init__(unit_system=unit_system_manager.current, **kwargs)
# pylint: disable=super-init-not-called
# pylint: disable=fixme
# TODO: avoid overloading the __init__ so IDE can proper prompt root level keys
def __init__(self, filename: str = None, file_content: dict = None, **kwargs):
if filename is not None or file_content is not None:
self._init_no_unit_context(filename, file_content, **kwargs)
else:
self._init_with_unit_context(**kwargs)
def copy(self, update=None, **kwargs) -> _ParamModelBase:
if unit_system_manager.current is None:
# pylint: disable=not-context-manager
with self.unit_system:
return super().copy(update=update, **kwargs)
return super().copy(update=update, **kwargs)
# pylint: disable=too-many-public-methods
[docs]
class SimulationParams(_ParamModelBase):
"""All-in-one class for surface meshing + volume meshing + case configurations"""
meshing: Optional[MeshingParams] = ConditionalField(
None,
context=[SURFACE_MESH, VOLUME_MESH],
description="Surface and volume meshing parameters. See :class:`MeshingParams` for more details.",
)
reference_geometry: Optional[ReferenceGeometry] = CaseField(
None,
description="Global geometric reference values. See :class:`ReferenceGeometry` for more details.",
)
operating_condition: Optional[OperatingConditionTypes] = CaseField(
None,
discriminator="type_name",
description="Global operating condition."
" See :ref:`Operating Condition <operating_condition>` for more details.",
)
#
# meshing->edge_refinement, face_refinement, zone_refinement, volumes and surfaces should be class which has the:
# 1. __getitem__ to allow [] access
# 2. __setitem__ to allow [] assignment
# 3. by_name(pattern:str) to use regexpr/glob to select all zones/surfaces with matched name
# 4. by_type(pattern:str) to use regexpr/glob to select all zones/surfaces with matched type
models: Optional[List[ModelTypes]] = CaseField(
None,
description="Solver settings and numerical models and boundary condition settings."
" See :ref:`Volume Models <volume_models>` and :ref:`Surface Models <surface_models>` for more details.",
)
time_stepping: Union[Steady, Unsteady] = CaseField(
Steady(),
discriminator="type_name",
description="Time stepping settings. See :ref:`Time Stepping <timeStepping>` for more details.",
)
user_defined_dynamics: Optional[List[UserDefinedDynamic]] = CaseField(
None,
description="User defined dynamics. See :ref:`User Defined Dynamics <user_defined_dynamics>` for more details.",
)
user_defined_fields: List[UserDefinedField] = CaseField(
[], description="User defined fields that can be used in outputs."
)
# Support for user defined expression?
# If so:
# 1. Move over the expression validation functions.
# 2. Have camelCase to snake_case naming converter for consistent user experience.
# Limitations:
# 1. No per volume zone output. (single volume output)
outputs: Optional[List[OutputTypes]] = CaseField(
None,
description="Output settings. See :ref:`Outputs <outputs>` for more details.",
)
run_control: Optional[RunControl] = CaseField(
None,
description="Run control settings of the simulation.",
)
##:: [INTERNAL USE ONLY] Private attributes that should not be modified manually.
private_attribute_asset_cache: AssetCache = pd.Field(AssetCache(), frozen=True)
private_attribute_dict: Optional[dict] = pd.Field(None)
# pylint: disable=arguments-differ
def _preprocess(self, mesh_unit=None, exclude: list = None) -> SimulationParams:
"""Internal function for non-dimensionalizing the simulation parameters"""
if exclude is None:
exclude = []
if mesh_unit is None:
raise Flow360ConfigurationError("Mesh unit has not been supplied.")
self._private_set_length_unit(LengthType.validate(mesh_unit)) # pylint: disable=no-member
if unit_system_manager.current is None:
# pylint: disable=not-context-manager
with self.unit_system:
return super().preprocess(
params=self, exclude=exclude, flow360_unit_system=self.flow360_unit_system
)
return super().preprocess(
params=self, exclude=exclude, flow360_unit_system=self.flow360_unit_system
)
def _private_set_length_unit(self, validated_mesh_unit):
with model_attribute_unlock(self.private_attribute_asset_cache, "project_length_unit"):
# pylint: disable=assigning-non-slot
self.private_attribute_asset_cache.project_length_unit = validated_mesh_unit
[docs]
@pd.validate_call
def convert_unit(
self,
value: DimensionedTypes,
target_system: Literal["SI", "Imperial", "flow360"],
length_unit: Optional[LengthType] = None,
):
"""
Converts a given value to the specified unit system.
This method takes a dimensioned quantity and converts it from its current unit system
to the target unit system, optionally considering a specific length unit for the conversion.
Parameters
----------
value : DimensionedTypes
The dimensioned quantity to convert. This should have units compatible with Flow360's
unit system.
target_system : str
The target unit system for conversion. Common values include "SI", "Imperial", "flow360".
length_unit : LengthType, optional
The length unit to use for conversion. If not provided, the method defaults to
the project length unit stored in the `private_attribute_asset_cache`.
Returns
-------
DimensionedTypes
The converted value in the specified target unit system.
Raises
------
Flow360RuntimeError
If the input unit system is not compatible with the target system, or if the required
length unit is missing.
Examples
--------
Convert a value from the current system to Flow360's V2 unit system:
>>> simulation_params = SimulationParams()
>>> value = unyt_quantity(1.0, "meters")
>>> converted_value = simulation_params.convert_unit(value, target_system="flow360")
>>> print(converted_value)
1.0 (flow360_length_unit)
"""
if length_unit is not None:
# pylint: disable=no-member
self._private_set_length_unit(LengthType.validate(length_unit))
flow360_conv_system = unit_converter(
value.units.dimensions,
params=self,
required_by=[f"{self.__class__.__name__}.convert_unit(value=, target_system=)"],
)
if target_system == "flow360":
target_system = "flow360_v2"
if is_flow360_unit(value) and not isinstance(value, unyt_quantity):
converted = value.in_base(target_system, flow360_conv_system)
else:
value.units.registry = flow360_conv_system.registry # pylint: disable=no-member
converted = value.in_base(unit_system=target_system)
return converted
# pylint: disable=no-self-argument
@pd.field_validator("models", mode="after")
@classmethod
def apply_default_fluid_settings(cls, v):
"""Apply default Fluid() settings if not found in mode`ls"""
if v is None:
v = []
assert isinstance(v, list)
if not any(isinstance(item, Fluid) for item in v):
v.append(Fluid(private_attribute_id="__default_fluid"))
return v
@pd.field_validator("models", mode="after")
@classmethod
def check_parent_volume_is_rotating(cls, models):
"""Ensure that all the parent volumes listed in the `Rotation` model are not static"""
return _check_parent_volume_is_rotating(models)
@pd.field_validator("models", mode="after")
@classmethod
def check_valid_models_for_liquid(cls, models):
"""Ensure that all the boundary conditions used are valid."""
return _check_valid_models_for_liquid(models)
@pd.field_validator("user_defined_dynamics", "user_defined_fields", mode="after")
@classmethod
def _disable_expression_for_liquid(cls, value, info: pd.ValidationInfo):
"""Ensure that string expressions are disabled for liquid simulation."""
validation_info = get_validation_info()
if validation_info is None or validation_info.using_liquid_as_material is False:
return value
if value:
raise ValueError(
f"{info.field_name} cannot be used when using liquid as simulation material."
)
return value
@pd.field_validator("outputs", mode="after")
@classmethod
def check_duplicate_isosurface_names(cls, outputs):
"""Check if we have isosurfaces with a duplicate name"""
return _check_duplicate_isosurface_names(outputs)
@pd.field_validator("outputs", mode="after")
@classmethod
def check_duplicate_surface_usage(cls, outputs):
"""Disallow the same boundary/surface being used in multiple outputs"""
return _check_duplicate_surface_usage(outputs)
@pd.field_validator("user_defined_fields", mode="after")
@classmethod
def check_duplicate_user_defined_fields(cls, v):
"""Check if we have duplicate user defined fields"""
if v == []:
return v
known_user_defined_fields = set()
for field in v:
if field.name in known_user_defined_fields:
raise ValueError(f"Duplicate user defined field name: {field.name}")
known_user_defined_fields.add(field.name)
return v
@pd.model_validator(mode="after")
def check_cht_solver_settings(self):
"""Check the Conjugate Heat Transfer settings, transferred from checkCHTSolverSettings"""
return _check_cht_solver_settings(self)
@pd.model_validator(mode="after")
def check_consistency_wall_function_and_surface_output(self):
"""Only allow wallFunctionMetric output field when there is a Wall model with a wall function enabled"""
return _check_consistency_wall_function_and_surface_output(self)
@pd.model_validator(mode="after")
def check_consistency_hybrid_model_volume_output(self):
"""Only allow hybrid RANS-LES output field when there is a corresponding solver with
hybrid RANS-LES enabled in models
"""
return _check_consistency_hybrid_model_volume_output(self)
@pd.model_validator(mode="after")
def check_unsteadiness_to_use_hybrid_model(self):
"""Only allow hybrid RANS-LES output field for unsteady simulations"""
return _check_unsteadiness_to_use_hybrid_model(self)
@pd.model_validator(mode="after")
def check_hybrid_model_to_use_zonal_enforcement(self):
"""Only allow LES/RANS zonal enforcement in hybrid RANS-LES mode"""
return _check_hybrid_model_to_use_zonal_enforcement(self)
@pd.model_validator(mode="after")
def check_unsteadiness_to_use_aero_acoustics(self):
"""Only allow Aero acoustics when using unsteady simulation"""
return _check_unsteadiness_to_use_aero_acoustics(self)
@pd.model_validator(mode="after")
def check_unique_surface_volume_probe_names(self):
"""Only allow unique probe names"""
return _check_unique_surface_volume_probe_names(self)
@pd.model_validator(mode="after")
def check_unique_surface_volume_probe_entity_names(self):
"""Only allow unique probe entity names"""
return _check_unique_surface_volume_probe_entity_names(self)
@pd.model_validator(mode="after")
def check_duplicate_entities_in_models(self):
"""Only allow each Surface/Volume entity to appear once in the Surface/Volume model"""
return _check_duplicate_entities_in_models(self)
@pd.model_validator(mode="after")
def check_numerical_dissipation_factor_output(self):
"""Only allow numericalDissipationFactor output field when the NS solver has low numerical dissipation"""
return _check_numerical_dissipation_factor_output(self)
@pd.model_validator(mode="after")
def check_low_mach_preconditioner_output(self):
"""Only allow lowMachPreconditioner output field when the lowMachPreconditioner is enabled in the NS solver"""
return _check_low_mach_preconditioner_output(self)
@pd.model_validator(mode="after")
@context_validator(context=CASE)
def check_complete_boundary_condition_and_unknown_surface(self):
"""Make sure that all boundaries have been assigned with a boundary condition"""
return _check_complete_boundary_condition_and_unknown_surface(self)
@pd.model_validator(mode="after")
def check_output_fields(params):
"""Check output fields and iso fields are valid"""
return _check_output_fields(params)
@pd.model_validator(mode="after")
def check_output_fields_valid_given_turbulence_model(params):
"""Check output fields are valid given the turbulence model"""
return _check_output_fields_valid_given_turbulence_model(params)
@pd.model_validator(mode="after")
def check_and_add_rotating_reference_frame_model_flag_in_volumezones(params):
"""Ensure that all volume zones have the rotating_reference_frame_model flag with correct values"""
return _check_and_add_noninertial_reference_frame_flag(params)
@pd.model_validator(mode="after")
def check_time_average_output(params):
"""Only allow TimeAverage output field in the unsteady simulations"""
return _check_time_average_output(params)
def _register_assigned_entities(self, registry: EntityRegistry) -> EntityRegistry:
"""Recursively register all entities listed in EntityList to the asset cache."""
# pylint: disable=no-member
registry.clear()
register_entity_list(self, registry)
return registry
def _update_entity_private_attrs(self, registry: EntityRegistry) -> EntityRegistry:
"""
Once the SimulationParams is set, extract and update information
into all used entities by parsing the params.
"""
##::1. Update full names in the Surface entities with zone names
# pylint: disable=no-member
if self.meshing is not None and self.meshing.volume_zones is not None:
for volume in self.meshing.volume_zones:
if isinstance(volume, AutomatedFarfield):
_set_boundary_full_name_with_zone_name(
registry,
"farfield",
volume.private_attribute_entity.name,
)
_set_boundary_full_name_with_zone_name(
registry,
"symmetric*",
volume.private_attribute_entity.name,
)
if isinstance(volume, (RotationCylinder, RotationVolume)):
# pylint: disable=fixme
# TODO: Implement this
pass
return registry
@property
def base_length(self) -> LengthType:
"""Get base length unit for non-dimensionalization"""
# pylint:disable=no-member
return self.private_attribute_asset_cache.project_length_unit.to("m")
@property
def base_temperature(self) -> AbsoluteTemperatureType:
"""Get base temperature unit for non-dimensionalization"""
# pylint:disable=no-member
if self.operating_condition.type_name == "LiquidOperatingCondition":
# Temperature in this condition has no effect because the thermal features will be disabled.
# Also the viscosity will be constant.
# pylint:disable = no-member
return 273 * u.K
return self.operating_condition.thermal_state.temperature.to("K")
@property
def base_velocity(self) -> VelocityType:
"""Get base velocity unit for non-dimensionalization"""
# pylint:disable=no-member
if self.operating_condition.type_name == "LiquidOperatingCondition":
# Provides an imaginary "speed of sound"
# Resulting in a hardcoded freestream mach of `LIQUID_IMAGINARY_FREESTREAM_MACH`
# To ensure incompressible range.
# pylint: disable=protected-access
if self.operating_condition._evaluated_velocity_magnitude.value != 0:
return (
self.operating_condition._evaluated_velocity_magnitude
/ LIQUID_IMAGINARY_FREESTREAM_MACH
).to("m/s")
return (
self.operating_condition.reference_velocity_magnitude # pylint:disable=no-member
/ LIQUID_IMAGINARY_FREESTREAM_MACH
).to("m/s")
return self.operating_condition.thermal_state.speed_of_sound.to("m/s")
@property
def _liquid_reference_velocity(self) -> VelocityType:
"""
This function returns the reference velocity for liquid operating condition.
Note that the reference velocity is **NOT** the non-dimensionalization velocity scale
For dimensionalization of Flow360 output (converting FROM flow360 unit)
The solver output is already re-normalized by `reference velocity` due to "velocityScale"
So we need to find the `reference velocity`.
`reference_velocity_magnitude` takes precedence, consistent with how "velocityScale" is computed.
"""
# pylint:disable=no-member
if self.operating_condition.reference_velocity_magnitude is not None:
reference_velocity = (self.operating_condition.reference_velocity_magnitude).to("m/s")
else:
reference_velocity = self.base_velocity.to("m/s") * LIQUID_IMAGINARY_FREESTREAM_MACH
return reference_velocity
@property
def base_density(self) -> DensityType:
"""Get base density unit for non-dimensionalization"""
# pylint:disable=no-member
if self.operating_condition.type_name == "LiquidOperatingCondition":
return self.operating_condition.material.density.to("kg/m**3")
return self.operating_condition.thermal_state.density.to("kg/m**3")
@property
def base_mass(self) -> MassType:
"""Get base mass unit for non-dimensionalization"""
return self.base_density * self.base_length**3
@property
def base_time(self) -> TimeType:
"""Get base time unit for non-dimensionalization"""
return self.base_length / self.base_velocity
@property
def flow360_unit_system(self) -> u.UnitSystem:
"""Get the unit system for non-dimensionalization"""
if self.operating_condition is None:
# Pure meshing mode
return u.UnitSystem(
name="flow360_nondim",
length_unit=self.base_length,
mass_unit=1 * u.kg, # pylint: disable=no-member
time_unit=1 * u.s, # pylint: disable=no-member
temperature_unit=1 * u.K, # pylint: disable=no-member
)
return u.UnitSystem(
name="flow360_nondim",
length_unit=self.base_length,
mass_unit=self.base_mass,
time_unit=self.base_time,
temperature_unit=self.base_temperature,
)
@property
def used_entity_registry(self) -> EntityRegistry:
"""
Get a entity registry that collects all the entities used in the simulation.
And also try to update the entities now that we have a global view of the simulation.
"""
registry = EntityRegistry()
registry = self._register_assigned_entities(registry)
registry = self._update_entity_private_attrs(registry)
return registry
def _update_param_with_actual_volume_mesh_meta(self, volume_mesh_meta_data: dict):
"""
Update the zone info from the actual volume mesh before solver execution.
Will be executed in the casePipeline as part of preprocessing.
Some thoughts:
Do we also need to update the params when the **surface meshing** is done?
"""
# pylint:disable=no-member
used_entity_registry = self.used_entity_registry
# Below includes the Ghost entities.
_update_entity_full_name(self, _SurfaceEntityBase, volume_mesh_meta_data)
_update_entity_full_name(self, _VolumeEntityBase, volume_mesh_meta_data)
_update_zone_boundaries_with_metadata(used_entity_registry, volume_mesh_meta_data)
return self
[docs]
def is_steady(self):
"""
returns True when SimulationParams is steady state
"""
return isinstance(self.time_stepping, Steady)
[docs]
def has_solid(self):
"""
returns True when SimulationParams has Solid model
"""
if self.models is None:
return False
# pylint: disable=not-an-iterable
return any(isinstance(item, Solid) for item in self.models)
[docs]
def has_actuator_disks(self):
"""
returns True when SimulationParams has ActuatorDisk disk
"""
if self.models is None:
return False
# pylint: disable=not-an-iterable
return any(isinstance(item, ActuatorDisk) for item in self.models)
[docs]
def has_bet_disks(self):
"""
returns True when SimulationParams has BET disk
"""
if self.models is None:
return False
# pylint: disable=not-an-iterable
return any(isinstance(item, BETDisk) for item in self.models)
[docs]
def has_isosurfaces(self):
"""
returns True when SimulationParams has isosurfaces
"""
if self.outputs is None:
return False
# pylint: disable=not-an-iterable
return any(isinstance(item, IsosurfaceOutput) for item in self.outputs)
[docs]
def has_monitors(self):
"""
returns True when SimulationParams has monitors
"""
if self.outputs is None:
return False
# pylint: disable=not-an-iterable
return any(
isinstance(item, (ProbeOutput, SurfaceProbeOutput, SurfaceIntegralOutput))
for item in self.outputs
)
[docs]
def has_volume_output(self):
"""
returns True when SimulationParams has volume output
"""
if self.outputs is None:
return False
# pylint: disable=not-an-iterable
return any(isinstance(item, VolumeOutput) for item in self.outputs)
[docs]
def has_aeroacoustics(self):
"""
returns True when SimulationParams has aeroacoustics
"""
if self.outputs is None:
return False
# pylint: disable=not-an-iterable
return any(isinstance(item, (AeroAcousticOutput)) for item in self.outputs)
[docs]
def has_user_defined_dynamics(self):
"""
returns True when SimulationParams has user defined dynamics
"""
return self.user_defined_dynamics is not None and len(self.user_defined_dynamics) > 0
[docs]
def display_output_units(self) -> None:
"""
Display all the output units for UserVariables used in `outputs`.
"""
if not self.outputs:
return
post_processing_variables = get_post_processing_variables(self)
# Sort for consistent behavior
post_processing_variables = sorted(post_processing_variables)
name_units_pair = batch_get_user_variable_units(post_processing_variables, self)
if not name_units_pair:
return
# Calculate column widths dynamically
name_column_width = max(len("Variable Name"), max(len(name) for name in name_units_pair))
unit_column_width = max(
len("Unit"), max(len(str(unit)) for unit in name_units_pair.values())
)
# Ensure minimum column widths
name_column_width = max(name_column_width, 15)
unit_column_width = max(unit_column_width, 10)
# Create the table header
header = f"{'Variable Name':<{name_column_width}} | {'Unit':<{unit_column_width}}"
separator = "-" * len(header)
# Print the table
log.info("")
log.info("Units of output `UserVariables`:")
log.info(separator)
log.info(header)
log.info(separator)
# Print each row
for name, unit in name_units_pair.items():
log.info(f"{name:<{name_column_width}} | {str(unit):<{unit_column_width}}")
log.info(separator)
log.info("")
[docs]
def pre_submit_summary(self):
"""
Display a summary of the simulation params before submission.
"""
self.display_output_units()
