user_defined_dynamics.py

"""User defined dynamic model for SimulationParams"""

from typing import Dict, List, Optional, Union

import pydantic as pd

from flow360.component.simulation.framework.base_model import Flow360BaseModel
from flow360.component.simulation.framework.entity_base import EntityList
from flow360.component.simulation.framework.expressions import StringExpression
from flow360.component.simulation.primitives import (
    CustomVolume,
    Cylinder,
    GenericVolume,
    MirroredSurface,
    SeedpointVolume,
    Surface,
)
from flow360.component.simulation.validation.validation_context import (
    ParamsValidationInfo,
    contextual_field_validator,
)
from flow360.component.simulation.validation.validation_utils import (
    validate_entity_list_surface_existence,
)


class UserDefinedDynamic(Flow360BaseModel):
    """
    :class:`UserDefinedDynamic` class for defining the user defined dynamics inputs.
    Please refer to :doc:`this example </python_api/example_library/notebooks/udd_alpha_controller>`
    for an implementation example.

    Example
    -------

    >>> fl.UserDefinedDynamic(
    ...    name="dynamicTheta",
    ...    input_vars=["momentY"],
    ...    constants={
    ...        "I": 0.443768309310345,
    ...        "zeta": zeta,
    ...        "K": K,
    ...        "omegaN": omegaN,
    ...        "theta0": theta0,
    ...    },
    ...    output_vars={
    ...        "omegaDot": "state[0];",
    ...        "omega": "state[1];",
    ...        "theta": "state[2];",
    ...    },
    ...    state_vars_initial_value=[str(initOmegaDot), "0.0", str(initTheta)],
    ...    update_law=[
    ...        "if (pseudoStep == 0) (momentY - K * ( state[2] - theta0 ) "
    ...         + "- 2 * zeta * omegaN * I *state[1] ) / I; else state[0];",
    ...        "if (pseudoStep == 0) state[1] + state[0] * timeStepSize; else state[1];",
    ...        "if (pseudoStep == 0) state[2] + state[1] * timeStepSize; else state[2];",
    ...    ],
    ...    input_boundary_patches=volume_mesh["plateBlock/noSlipWall"],
    ...    output_target=volume_mesh["plateBlock"],
    ... )

    ====

    """

    name: str = pd.Field(
        "User defined dynamics", description="Name of the dynamics defined by the user."
    )
    input_vars: List[str] = pd.Field(
        description="List of the inputs to define the user defined dynamics. Inputs can be: :code:`CL`, :code:`CD`, "
        + ":code:`bet_NUM_torque`,  :code:`bet_NUM_thrust`, (NUM is the index of the BET disk starting from 0), "
        + ":code:`momentX`, :code:`momentY`, :code:`momentZ` (X/Y/Z moments with respect to "
        + ":py:attr:`~ReferenceGeometry.moment_center`), :code:`forceX`, :code:`forceY`, :code:`forceZ`. "
    )
    constants: Optional[Dict[str, float]] = pd.Field(
        None, description="A list of constants that can be used in the expressions."
    )
    output_vars: Optional[Dict[str, StringExpression]] = pd.Field(
        None,
        description="Name of the output variables and the expression for the output variables using state "
        + "variables. Outputs can be: :code:`alphaAngle`, :code:`betaAngle`, "
        + ":code:`bet_NUM_omega` (NUM is the index of the BET disk starting from 0), "
        + ":code:`theta`, :code:`omega` and :code:`omegaDot` (rotation angle/velocity/acceleration "
        + "in radians for sliding interfaces), :code:`actuatorDisk_<DISK_ENTITY_NAME>_thrustMultiplier` "
        + "and :code:`actuatorDisk_<DISK_ENTITY_NAME>_torqueMultiplier` (where <DISK_ENTITY_NAME> is "
        + "the name of the actuator disk entity). Please exercise caution when choosing output "
        + "variables, as any modifications to their values will be directly mirrored in the solver.",
    )
    state_vars_initial_value: List[StringExpression] = pd.Field(
        description="The initial value of state variables are specified here. The entries could be either values "
        + "(in the form of strings, e.g., :code:`0.0`) or expression with constants defined earlier or any input "
        + "and output variable. (e.g., :code:`2.0 * alphaAngle + someConstant`). The list entries correspond to "
        + "the initial values for :code:`state[0]`, :code:`state[1]`, ..., respectively."
    )
    update_law: List[StringExpression] = pd.Field(
        "List of expressions for updating state variables. The list entries correspond to the update laws for "
        + ":code:`state[0]`, :code:`state[1]`, ..., respectively. These expressions follows similar guidelines as "
        + ":ref:`user Defined Expressions<UserDefinedExpressions>`."
    )
    input_boundary_patches: Optional[EntityList[Surface, MirroredSurface]] = pd.Field(
        None,
        description="The list of :class:`~flow360.Surface` entities to which the input variables belongs. "
        + "If multiple boundaries are specified then the summation over the boundaries are used as the input. "
        + "For input variables that already specified the source in the name (like bet_NUM_torque) "
        + "this entry does not have any effect.",
    )
    output_target: Optional[Union[Cylinder, GenericVolume, Surface, CustomVolume]] = pd.Field(
        None,
        description="The target to which the output variables belong to. For example this can be the rotating "
        + "volume zone name. Only one output target is supported per user defined dynamics instance. Only "
        + ":class:`~flow360.Cylinder` entity is supported as target for now.",
    )  # Limited to `Cylinder` for now as we have only tested using UDD to control rotation.

    @contextual_field_validator("input_boundary_patches", mode="after")
    @classmethod
    def ensure_surface_existence(cls, value, param_info: ParamsValidationInfo):
        """Ensure all boundaries will be present after mesher"""
        return validate_entity_list_surface_existence(value, param_info)

    @contextual_field_validator("output_target", mode="after")
    @classmethod
    def ensure_output_surface_existence(cls, value, param_info: ParamsValidationInfo):
        """Ensure that the output target surface is not a deleted surface"""

        # pylint: disable=fixme, duplicate-code
        # TODO: We can make this a Surface's after model validator once entity info is separated from params.
        # TODO: And therefore no need for duplicate-code override.
        # pylint: disable=protected-access
        if isinstance(value, Surface) and value._will_be_deleted_by_mesher(
            entity_transformation_detected=param_info.entity_transformation_detected,
            farfield_method=param_info.farfield_method,
            global_bounding_box=param_info.global_bounding_box,
            planar_face_tolerance=param_info.planar_face_tolerance,
            half_model_symmetry_plane_center_y=param_info.half_model_symmetry_plane_center_y,
            quasi_3d_symmetry_planes_center_y=param_info.quasi_3d_symmetry_planes_center_y,
            farfield_domain_type=param_info.farfield_domain_type,
        ):
            raise ValueError(
                f"Boundary `{value.name}` will likely be deleted after mesh generation. Therefore it cannot be used."
            )
        return value

    @contextual_field_validator("output_target", mode="after")
    @classmethod
    def _ensure_custom_volume_is_valid(
        cls,
        value: Optional[Union[GenericVolume, Cylinder, CustomVolume, SeedpointVolume]],
        param_info: ParamsValidationInfo,
    ):
        """Ensure parent volume is a custom volume."""
        if value is None:
            return value
        if not isinstance(value, (CustomVolume, SeedpointVolume)):
            return value
        if value.name not in param_info.to_be_generated_custom_volumes:
            raise ValueError(
                f"Parent {type(value).__name__} {value.name} is not listed under meshing->volume_zones(or zones)"
                + "->CustomZones."
            )
        return value