Add service functions for translations.

flow360/component/simulation/meshing_param/edge_params.py

@@ -50,6 +50,9 @@ class SurfaceEdgeRefinement(_BaseEdgeRefinement):
     (equivalent to `ProjectAniso` in old params).
     """
 
+    refinement_type: Literal["SurfaceEdgeRefinement"] = pd.Field(
+        "SurfaceEdgeRefinement", frozen=True
+    )
     method: Optional[
         Union[
             AngleBasedRefinement,

flow360/component/simulation/meshing_param/face_params.py

@@ -25,6 +25,7 @@ class SurfaceRefinement(Flow360BaseModel):
     these defaults so that the when face name is not present, what config we ues. Depending on how we go down the road.
     """
 
+    refinement_type: Literal["SurfaceRefinement"] = pd.Field("SurfaceRefinement", frozen=True)
     entities: Optional[EntityList[Surface]] = pd.Field(None, alias="faces")
     max_edge_length: LengthType.Positive = pd.Field(
         description="Local maximum edge length for surface cells."
@@ -50,6 +51,7 @@ class BoundaryLayer(Flow360BaseModel):
     have dedicated field for global settings. This is also consistent with the `FluidDynamics` class' design.
     """
 
+    refinement_type: Literal["BoundaryLayer"] = pd.Field("BoundaryLayer", frozen=True)
     type: Literal["aniso", "projectAnisoSpacing", "none"] = pd.Field()
     entities: Optional[EntityList[Surface]] = pd.Field(None, alias="faces")
     first_layer_thickness: LengthType.Positive = pd.Field(

flow360/component/simulation/meshing_param/params.py

@@ -1,4 +1,4 @@
-from typing import List, Literal, Optional, Union
+from typing import Annotated, List, Literal, Optional, Union
 
 import pydantic as pd
 
@@ -12,6 +12,11 @@
     VolumeRefinementTypes,
 )
 
+AllowedRefinementTypes = Annotated[
+    Union[SurfaceEdgeRefinement, SurfaceRefinementTypes, VolumeRefinementTypes],
+    pd.Field(discriminator="refinement_type"),
+]
+
 
 class MeshingParams(Flow360BaseModel):
     """
@@ -57,9 +62,7 @@ class MeshingParams(Flow360BaseModel):
         None, ge=1, description="Global growth rate of the anisotropic layers grown from the edges."
     )  # Conditionally optional
 
-    refinements: Optional[
-        List[Union[SurfaceEdgeRefinement, SurfaceRefinementTypes, VolumeRefinementTypes]]
-    ] = pd.Field(
+    refinements: Optional[List[AllowedRefinementTypes]] = pd.Field(
         None,
         description="Additional fine-tunning for refinements.",
     )  # Note: May need discriminator for performance??

flow360/component/simulation/meshing_param/volume_params.py

@@ -1,4 +1,4 @@
-from typing import List, Literal, Optional, Union
+from typing import Literal, Optional, Union
 
 import pydantic as pd
 
@@ -13,6 +13,7 @@
 
 
 class UniformRefinement(Flow360BaseModel):
+    refinement_type: Literal["UniformRefinement"] = pd.Field("UniformRefinement", frozen=True)
     entities: EntityList[Box, Cylinder] = pd.Field()
     spacing: LengthType.Positive = pd.Field()
 
@@ -27,6 +28,9 @@ class AxisymmetricRefinement(Flow360BaseModel):
     - We may provide a helper function to automatically determine what is inside the encloeud_objects list based on the mesh data. But this currently is out of scope due to the estimated efforts.
     """
 
+    refinement_type: Literal["AxisymmetricRefinement"] = pd.Field(
+        "AxisymmetricRefinement", frozen=True
+    )
     entities: EntityList[Cylinder] = pd.Field()
     spacing_axial: LengthType.Positive = pd.Field()
     spacing_radial: LengthType.Positive = pd.Field()

flow360/component/simulation/outputs/outputs.py

@@ -1,4 +1,4 @@
-from typing import List, Literal, Optional, Tuple, Union
+from typing import Annotated, List, Literal, Union
 
 import pydantic as pd
 
@@ -62,6 +62,7 @@ class SurfaceOutput(_AnimationAndFileFormatSettings):
         description="Enable writing all surface outputs into a single file instead of one file per surface. This option currently only supports Tecplot output format. Will choose the value of the last instance of this option of the same output type (SurfaceOutput or TimeAverageSurfaceOutput) in the `output` list.",
     )
     output_fields: UniqueAliasedStringList[SurfaceFieldNames] = pd.Field()
+    output_type: Literal["SurfaceOutput"] = pd.Field("SurfaceOutput", frozen=True)
 
 
 class TimeAverageSurfaceOutput(SurfaceOutput):
@@ -77,10 +78,14 @@ class TimeAverageSurfaceOutput(SurfaceOutput):
     start_step: Union[pd.NonNegativeInt, Literal[-1]] = pd.Field(
         default=-1, description="Physical time step to start calculating averaging"
     )
+    output_type: Literal["TimeAverageSurfaceOutput"] = pd.Field(
+        "TimeAverageSurfaceOutput", frozen=True
+    )
 
 
 class VolumeOutput(_AnimationAndFileFormatSettings):
     output_fields: UniqueAliasedStringList[VolumeFieldNames] = pd.Field()
+    output_type: Literal["VolumeOutput"] = pd.Field("VolumeOutput", frozen=True)
 
 
 class TimeAverageVolumeOutput(VolumeOutput):
@@ -97,32 +102,40 @@ class TimeAverageVolumeOutput(VolumeOutput):
     start_step: Union[pd.NonNegativeInt, Literal[-1]] = pd.Field(
         default=-1, description="Physical time step to start calculating averaging"
     )
+    output_type: Literal["TimeAverageVolumeOutput"] = pd.Field(
+        "TimeAverageVolumeOutput", frozen=True
+    )
 
 
 class SliceOutput(_AnimationAndFileFormatSettings):
     entities: UniqueItemList[Slice] = pd.Field(alias="slices")
     output_fields: UniqueAliasedStringList[SliceFieldNames] = pd.Field()
+    output_type: Literal["SliceOutput"] = pd.Field("SliceOutput", frozen=True)
 
 
 class IsosurfaceOutput(_AnimationAndFileFormatSettings):
     entities: UniqueItemList[Isosurface] = pd.Field(alias="isosurfaces")
     output_fields: UniqueAliasedStringList[CommonFieldNames] = pd.Field()
+    output_type: Literal["IsosurfaceOutput"] = pd.Field("IsosurfaceOutput", frozen=True)
 
 
 class SurfaceIntegralOutput(_AnimationSettings):
     entities: UniqueItemList[SurfaceList] = pd.Field(alias="monitors")
     output_fields: UniqueAliasedStringList[CommonFieldNames] = pd.Field()
+    output_type: Literal["SurfaceIntegralOutput"] = pd.Field("SurfaceIntegralOutput", frozen=True)
 
 
 class ProbeOutput(_AnimationSettings):
     entities: UniqueItemList[Probe] = pd.Field(alias="probes")
     output_fields: UniqueAliasedStringList[CommonFieldNames] = pd.Field()
+    output_type: Literal["ProbeOutput"] = pd.Field("ProbeOutput", frozen=True)
 
 
 class AeroAcousticOutput(Flow360BaseModel):
     patch_type: str = pd.Field("solid", frozen=True)
     observers: List[LengthType.Point] = pd.Field()
     write_per_surface_output: bool = pd.Field(False)
+    output_type: Literal["AeroAcousticOutput"] = pd.Field("AeroAcousticOutput", frozen=True)
 
 
 class UserDefinedFields(Flow360BaseModel):
@@ -131,14 +144,17 @@ class UserDefinedFields(Flow360BaseModel):
     pass
 
 
-OutputTypes = Union[
-    SurfaceOutput,
-    TimeAverageSurfaceOutput,
-    VolumeOutput,
-    TimeAverageVolumeOutput,
-    SliceOutput,
-    IsosurfaceOutput,
-    SurfaceIntegralOutput,
-    ProbeOutput,
-    AeroAcousticOutput,
+OutputTypes = Annotated[
+    Union[
+        SurfaceOutput,
+        TimeAverageSurfaceOutput,
+        VolumeOutput,
+        TimeAverageVolumeOutput,
+        SliceOutput,
+        IsosurfaceOutput,
+        SurfaceIntegralOutput,
+        ProbeOutput,
+        AeroAcousticOutput,
+    ],
+    pd.Field(discriminator="output_type"),
 ]

flow360/component/simulation/services.py

@@ -4,6 +4,13 @@
 import pydantic as pd
 
 from flow360.component.simulation.simulation_params import SimulationParams
+from flow360.component.simulation.translator.solver_translator import get_solver_json
+from flow360.component.simulation.translator.surface_meshing_translator import (
+    get_surface_meshing_json,
+)
+from flow360.component.simulation.translator.volume_meshing_translator import (
+    get_volume_meshing_json,
+)
 from flow360.component.simulation.unit_system import (
     CGS_unit_system,
     SI_unit_system,
@@ -12,6 +19,7 @@
     imperial_unit_system,
     unit_system_manager,
 )
+from flow360.log import log
 
 unit_system_map = {
     "SI": SI_unit_system,
@@ -65,10 +73,11 @@ def validate_model(params_as_dict, unit_system_name):
     unit_system = init_unit_system(unit_system_name)
 
     validation_errors = None
+    validated_param = None
 
     try:
         with unit_system:
-            SimulationParams(**params_as_dict)
+            validated_param = SimulationParams(**params_as_dict)
     except pd.ValidationError as err:
         validation_errors = err.errors()
         # We do not care about handling / propagating the validation errors here,
@@ -92,4 +101,75 @@ def validate_model(params_as_dict, unit_system_name):
                     errors_as_list.remove(field)
                     error["loc"] = tuple(errors_as_list)
 
-    return validation_errors
+    return validated_param, validation_errors
+
+
+# pylint: disable=too-many-arguments
+def _translate_simulation_json(
+    params_as_dict,
+    unit_system_name,
+    mesh_unit,
+    existing_json=None,
+    target_name: str = None,
+    translation_func=None,
+):
+    """
+    Get JSON for surface meshing from a given simulaiton JSON.
+
+    """
+    translated_dict = None
+    param, errors = validate_model(params_as_dict, unit_system_name)
+    if errors is not None:
+        # pylint: disable=fixme
+        # TODO: Check if this looks good in terminal.
+        raise ValueError(errors)
+    if mesh_unit is None:
+        raise ValueError("Mesh unit is required for translation.")
+    try:
+        translated_dict = translation_func(param, mesh_unit)
+    except Exception as err:  # tranlsation itself is not supposed to raise any exception
+        raise ValueError(f"Failed to translate to {target_name} json: " + str(err)) from err
+    if translated_dict == {}:
+        raise ValueError(f"No {target_name} parameters found in given SimulationParams.")
+    # pylint: disable=fixme
+    # TODO: Implement proper hashing. Currently floating point creates headache for reproducible hashing.
+    if existing_json is not None and hash(translated_dict) == existing_json["hash"]:
+        log.info(f"Translation of {target_name} is same as existing. Skipping translation.")
+        return existing_json
+    return translated_dict
+
+
+def simulation_to_surface_meshing_json(params_as_dict, unit_system_name, mesh_unit, existing_json):
+    """Get JSON for surface meshing from a given simulaiton JSON."""
+    return _translate_simulation_json(
+        params_as_dict,
+        unit_system_name,
+        mesh_unit,
+        existing_json,
+        "surface meshing",
+        get_surface_meshing_json,
+    )
+
+
+def simulation_to_volume_meshing_json(params_as_dict, unit_system_name, mesh_unit, existing_json):
+    """Get JSON for volume meshing from a given simulaiton JSON."""
+    return _translate_simulation_json(
+        params_as_dict,
+        unit_system_name,
+        mesh_unit,
+        existing_json,
+        "volume meshing",
+        get_volume_meshing_json,
+    )
+
+
+def simulation_to_case_json(params_as_dict, unit_system_name, mesh_unit, existing_json):
+    """Get JSON for case from a given simulaiton JSON."""
+    return _translate_simulation_json(
+        params_as_dict,
+        unit_system_name,
+        mesh_unit,
+        existing_json,
+        "case",
+        get_solver_json,
+    )

flow360/component/simulation/translator/utils.py

@@ -6,18 +6,22 @@
 from flow360.component.simulation.simulation_params import (
     SimulationParams,  # Not required
 )
+from flow360.component.simulation.unit_system import LengthType
 
 
 def preprocess_input(func):
     @functools.wraps(func)
     def wrapper(input_params, mesh_unit, *args, **kwargs):
-        processed_input = get_simulation_param_dict(input_params, mesh_unit)
-        return func(processed_input, mesh_unit, *args, **kwargs)
+        validated_mesh_unit = LengthType.validate(mesh_unit)
+        processed_input = get_simulation_param_dict(input_params, validated_mesh_unit)
+        return func(processed_input, validated_mesh_unit, *args, **kwargs)
 
     return wrapper
 
 
-def get_simulation_param_dict(input_params: SimulationParams | str | dict, mesh_unit):
+def get_simulation_param_dict(
+    input_params: SimulationParams | str | dict, validated_mesh_unit: LengthType
+):
     """
     Get the dictionary of `SimulationParams`.
     """
@@ -40,7 +44,7 @@ def get_simulation_param_dict(input_params: SimulationParams | str | dict, mesh_
         param = SimulationParams(**input_params)
 
     if param is not None:
-        return param.preprocess(mesh_unit)
+        return param.preprocess(validated_mesh_unit)
     raise ValueError(f"Invalid input <{input_params.__class__.__name__}> for translator. ")
 
 

tests/simulation/service/test_services_v2.py

@@ -60,7 +60,7 @@ def test_validate_service():
         "user_defined_dynamics": [],
     }
 
-    errors = services.validate_model(params_as_dict=params_data, unit_system_name="SI")
+    _, errors = services.validate_model(params_as_dict=params_data, unit_system_name="SI")
 
     assert errors is None
 
@@ -117,7 +117,7 @@ def test_validate_error():
         "user_defined_dynamics": [],
     }
 
-    errors = services.validate_model(params_as_dict=params_data, unit_system_name="SI")
+    _, errors = services.validate_model(params_as_dict=params_data, unit_system_name="SI")
 
     assert len(errors) == 1
     assert errors[0]["loc"] == ("meshing", "farfield")
@@ -175,7 +175,7 @@ def test_validate_multiple_errors():
         "user_defined_dynamics": [],
     }
 
-    errors = services.validate_model(params_as_dict=params_data, unit_system_name="SI")
+    _, errors = services.validate_model(params_as_dict=params_data, unit_system_name="SI")
 
     assert len(errors) == 2
     assert errors[0]["loc"] == ("meshing", "farfield")