Feat/functions

src/ansys/fluent/core/solver/function/__init__.py

@@ -0,0 +1 @@
+"""Module providing reductions functions."""

src/ansys/fluent/core/solver/function/reduction.py

@@ -0,0 +1,299 @@
+"""Module providing reductions functions that can be applied to Fluent data
+from one or across multiple remote Fluent sessions.
+
+The following parameters are relevant for the reduction functions. The
+expr parameter is not relevant to all reductions functions.
+
+Parameters
+----------
+expr : Any
+    Expression that can be either a string or an
+    instance of a specific settings API named_expressions
+    object. The expression can be a field variable or a
+    a valid Fluent expression. A specified named expression
+    can be handled for multiple solvers as long as the
+    expression's definition is valid in each solver (it
+    does not need to be created in each solver)
+locations : Any
+    A list of location strings, or an API object that can be
+    resolved to a list of location strings
+    (e.g., setup.boundary_conditions,
+    or results.surfaces.plane_surface),
+    or a list of such objects. If location strings are
+    included in the list, then only string must be included
+ctxt : Any, optional
+    An optional API object (e.g., the root solver session
+    object but any solver API object will suffice) to set
+    the context of the call's execution. If the location
+    objects are strings, then such a context is required
+Returns
+-------
+float
+    The result of the reduction
+
+Examples
+--------
+
+>>> from ansys.fluent.core.solver.function import reduction
+>>> # Compute the area average of absolute pressure across all boundary
+>>> # condition surfaces of the given solver
+>>> reduction.area_average(
+...     expr = "AbsolutePressure",
+...     locations = solver.setup.boundary_conditions.velocity_inlet
+... )
+10623.0
+
+>>> from ansys.fluent.core.solver.function import reduction
+>>> # Compute the minimum of the square of velocity magnitude
+>>> # for all pressure outlets across two solvers
+>>> named_exprs = solver1.setup.named_expressions
+>>> vsquared = named_exprs["vsquared"] = {}
+>>> vsquared.definition = "VelocityMagnitude ** 2"
+>>> reduction.minimum(
+...     expr = vsquared,
+...     locations = [
+...         solver1.setup.boundary_conditions.pressure_outlet,
+...         solver2.setup.boundary_conditions.pressure_outlet
+...     ])
+19.28151
+
+>>> from ansys.fluent.core.solver.function import reduction
+>>> # Compute the minimum of the square of velocity magnitude
+>>> # for all pressure outlets across two solvers
+>>> named_exprs = solver1.setup.named_expressions
+>>> vsquared = named_exprs["vsquared"] = {}
+>>> vsquared.definition = "VelocityMagnitude ** 2"
+>>> reduction.find_minimum(
+...     expr = vsquared,
+...     locations = [
+...         solver1.setup.boundary_conditions.pressure_outlet,
+...         solver2.setup.boundary_conditions.pressure_outlet
+...     ])
+[('session-1', 'outlet1')]
+"""
+
+
+class BadReductionRequest(Exception):
+    def __init__(self, err):
+        super().__init__(f"Could not complete reduction function request: {err}")
+
+
+def _is_iterable(obj):
+    return hasattr(type(obj), "__iter__")
+
+
+def _expand_locn_container(locns):
+    try:
+        return [[locn, locns] for locn in locns]
+    except TypeError as ex:
+        raise BadReductionRequest(ex)
+
+
+def _locn_name_and_obj(locn, locns):
+    if isinstance(locn, str):
+        return [locn, locns]
+    # should call locn_get_name()
+    if _is_iterable(locn):
+        return _locn_names_and_objs(locn)
+    else:
+        return [locn.obj_name, locn]
+
+
+def _locn_names_and_objs(locns):
+    if _is_iterable(locns):
+        names_and_objs = []
+        for locn in locns:
+            name_and_obj = _locn_name_and_obj(locn, locns)
+            if _is_iterable(name_and_obj):
+                if isinstance(name_and_obj[0], str):
+                    names_and_objs.append(name_and_obj)
+                else:
+                    names_and_objs.extend(name_and_obj)
+        return names_and_objs
+    else:
+        return _expand_locn_container(locns)
+
+
+def _root(obj):
+    return obj if not getattr(obj, "obj_name", None) else _root(obj._parent)
+
+
+def _validate_locn_list(locn_list, ctxt):
+    if not all(locn[0] for locn in locn_list) and (
+        any(locn[0] for locn in locn_list) or not ctxt
+    ):
+        raise BadReductionRequest("Invalid combination of arguments")
+
+
+def _locns(locns, ctxt):
+    locn_names_and_objs = _locn_names_and_objs(locns)
+    locn_list = []
+    for name, obj in locn_names_and_objs:
+        root = _root(obj)
+        found = False
+        for locn in locn_list:
+            if locn[0] is root:
+                locn[1].append(name)
+                found = True
+                break
+        if not found:
+            locn_list.append([root, [name]])
+    _validate_locn_list(locn_list, ctxt)
+    return locn_list
+
+
+def _eval_expr(solver, expr_str):
+    named_exprs = solver.setup.named_expressions
+    expr_name = "temp_expr_1"
+    named_exprs[expr_name] = {}
+    # request feature: anonymous name object creation
+    expr_obj = named_exprs["temp_expr_1"]
+    expr_obj.definition = expr_str
+    val = expr_obj.get_value()
+    named_exprs.pop(expr_name)
+    return val
+
+
+def _expr_to_expr_str(expr):
+    return getattr(expr, "definition", expr) if expr is not None else expr
+
+
+def _eval_reduction(solver, reduction, locations, expr=None):
+    expr_str = _expr_to_expr_str(expr)
+    return _eval_expr(
+        solver,
+        (
+            f"{reduction}({locations})"
+            if expr_str is None
+            else f"{reduction}({expr_str},{locations})"
+        ),
+    )
+
+
+def _extent_average(extent_name, expr, locations, ctxt):
+    locns = _locns(locations, ctxt)
+    numerator = 0.0
+    denominator = 0.0
+    for solver, names in locns:
+        solver = solver or _root(ctxt)
+        val = _eval_reduction(solver, f"{extent_name}Ave", names, expr)
+        extent = _eval_reduction(solver, extent_name, names) if len(locns) > 1 else 1
+        numerator += val * extent
+        denominator += extent
+    if denominator == 0.0:
+        raise BadReductionRequest("Zero extent computed for average")
+    return numerator / denominator
+
+
+def _extent(extent_name, locations, ctxt):
+    locns = _locns(locations, ctxt)
+    total = 0.0
+    for solver, names in locns:
+        solver = solver or _root(ctxt)
+        extent = _eval_expr(solver, f"{extent_name}({names})")
+        total += extent
+    return total
+
+
+def _limit(limit, expr, locations, ctxt):
+    locns = _locns(locations, ctxt)
+    limit_val = None
+    for solver, names in locns:
+        solver = solver or _root(ctxt)
+        val = _eval_reduction(
+            solver, "Minimum" if limit is min else "Maximum", names, expr
+        )
+        limit_val = val if limit_val is None else limit(val, limit_val)
+    return limit_val
+
+
+def area_average(expr, locations, ctxt=None):
+    """Compute the area average of the specified expression over the specified
+    locations.
+
+    Parameters
+    ----------
+    expr : Any
+    locations : Any
+    ctxt : Any, optional
+    Returns
+    -------
+    float
+    """
+    return _extent_average("Area", expr, locations, ctxt)
+
+
+def volume_average(expr, locations, ctxt=None):
+    """Compute the volume average of the specified expression over the
+    specified locations.
+
+    Parameters
+    ----------
+    expr : Any
+    locations : Any
+    ctxt : Any, optional
+    Returns
+    -------
+    float
+    """
+    return _extent_average("Volume", expr, locations, ctxt)
+
+
+def area(locations, ctxt=None):
+    """Compute the total area of the specified locations.
+
+    Parameters
+    ----------
+    locations : Any
+    ctxt : Any, optional
+    Returns
+    -------
+    float
+    """
+    return _extent("Area", locations, ctxt)
+
+
+def volume(locations, ctxt=None):
+    """Compute the total volume of the specified locations.
+
+    Parameters
+    ----------
+    locations : Any
+    ctxt : Any, optional
+    Returns
+    -------
+    float
+    """
+    return _extent("Volume", locations, ctxt)
+
+
+def minimum(expr, locations, ctxt=None):
+    """Compute the minimum of the specified expression over the specified
+    locations.
+
+    Parameters
+    ----------
+    expr : Any
+    locations : Any
+    ctxt : Any, optional
+    Returns
+    -------
+    float
+    """
+    return _limit(min, expr, locations, ctxt)
+
+
+def maximum(expr, locations, ctxt=None):
+    """Compute the maximum of the specified expression over the specified
+    locations.
+
+    Parameters
+    ----------
+    expr : Any
+    locations : Any
+    ctxt : Any, optional
+    Returns
+    -------
+    float
+    """
+    return _limit(max, expr, locations, ctxt)

tests/test_reduction.py

@@ -0,0 +1,90 @@
+import pytest
+from util.fixture_fluent import load_static_mixer_case  # noqa: F401
+
+from ansys.fluent.core.solver.function import reduction
+
+load_static_mixer_case_2 = load_static_mixer_case
+
+
+def _test_locn_extraction(solver1, solver2):
+    locns = reduction._locn_names_and_objs(["inlet1"])
+    assert locns == [["inlet1", ["inlet1"]]]
+
+    all_bcs = solver1.setup.boundary_conditions
+    locns = reduction._locn_names_and_objs(all_bcs)
+    assert locns == [
+        ["interior--fluid", all_bcs],
+        ["outlet", all_bcs],
+        ["inlet1", all_bcs],
+        ["inlet2", all_bcs],
+        ["wall", all_bcs],
+    ]
+
+    locns = reduction._locn_names_and_objs([all_bcs["inlet1"]])
+    assert locns == [["inlet1", all_bcs["inlet1"]]]
+
+    all_bcs = solver1.setup.boundary_conditions
+    all_bcs2 = solver2.setup.boundary_conditions
+    locns = reduction._locn_names_and_objs([all_bcs, all_bcs2])
+    assert locns == [
+        ["interior--fluid", all_bcs],
+        ["outlet", all_bcs],
+        ["inlet1", all_bcs],
+        ["inlet2", all_bcs],
+        ["wall", all_bcs],
+        ["interior--fluid", all_bcs2],
+        ["outlet", all_bcs2],
+        ["inlet1", all_bcs2],
+        ["inlet2", all_bcs2],
+        ["wall", all_bcs2],
+    ]
+
+
+def _test_area_average(solver):
+    solver.solution.initialization.hybrid_initialize()
+    solver.setup.named_expressions["test_expr_1"] = {}
+    solver.setup.named_expressions[
+        "test_expr_1"
+    ].definition = "AreaAve(AbsolutePressure, ['inlet1'])"
+    expr_val = solver.setup.named_expressions["test_expr_1"].get_value()
+    assert type(expr_val) == float and expr_val != 0.0
+    val = reduction.area_average(
+        expr="AbsolutePressure",
+        locations=solver.setup.boundary_conditions.velocity_inlet,
+    )
+    assert val == expr_val
+    solver.setup.named_expressions.pop(key="test_expr_1")
+
+
+def _test_min(solver1, solver2):
+    vmag = solver1.setup.boundary_conditions["inlet1"].vmag.value()
+    solver1.setup.boundary_conditions["inlet1"].vmag = 0.9 * vmag
+    solver2.setup.boundary_conditions["inlet1"].vmag = 1.1 * vmag
+    solver1.solution.initialization.hybrid_initialize()
+    solver2.solution.initialization.hybrid_initialize()
+    solver1.setup.named_expressions["test_expr_1"] = {}
+    test_expr1 = solver1.setup.named_expressions["test_expr_1"]
+    test_expr1.definition = "sqrt(VelocityMagnitude)"
+    solver1.setup.named_expressions["test_expr_2"] = {}
+    test_expr2 = solver1.setup.named_expressions["test_expr_2"]
+    test_expr2.definition = "minimum(test_expr_2, ['outlet'])"
+    expected_result = test_expr2.get_value()
+    result = reduction.minimum(
+        test_expr1,
+        [
+            solver1.setup.boundary_conditions["outlet"],
+            solver2.setup.boundary_conditions["outlet"],
+        ],
+    )
+    assert result == expected_result
+    solver1.setup.named_expressions.pop(key="test_expr_1")
+    solver1.setup.named_expressions.pop(key="test_expr_2")
+
+
+@pytest.mark.fluent_231
+def test_reductions(load_static_mixer_case, load_static_mixer_case_2) -> None:
+    solver1 = load_static_mixer_case
+    solver2 = load_static_mixer_case_2
+    _test_locn_extraction(solver1, solver2)
+    _test_area_average(solver1)
+    _test_min(solver1, solver2)