.. DO NOT EDIT.
.. THIS FILE WAS AUTOMATICALLY GENERATED BY SPHINX-GALLERY.
.. TO MAKE CHANGES, EDIT THE SOURCE PYTHON FILE:
.. "auto_how_to/example_custom_aggregator.py"
.. LINE NUMBERS ARE GIVEN BELOW.

.. only:: html

    .. note::
        :class: sphx-glr-download-link-note

        :ref:`Go to the end <sphx_glr_download_auto_how_to_example_custom_aggregator.py>`
        to download the full example code.

.. rst-class:: sphx-glr-example-title

.. _sphx_glr_auto_how_to_example_custom_aggregator.py:


Write a custom aggregator node
==============================

Aggregators are the "reduce to a number" step of a pipeline. The built-in
ones (``Sum``, ``Mean``, ``Min``, ``Max``, ``VolIntegrate``, ``SurfIntegrate``)
cover most cases, but sometimes you need a specific reduction. Here we
build a ``Variance`` node from scratch, verify it matches a numpy
reference, and slot it into the pipeline.

Details on the node contract live in :doc:`/explanation/custom_nodes` —
this page is the practical walk-through.

.. GENERATED FROM PYTHON SOURCE LINES 16-18

Open the case
-------------

.. GENERATED FROM PYTHON SOURCE LINES 18-41

.. code-block:: Python


    import os
    import subprocess

    import pyOFTools.patch_pybfoam  # noqa: F401
    from pyOFTools import clone_example

    CASE = clone_example("damBreak")
    subprocess.run(
        ["./Allrun"],
        cwd=CASE,
        check=True,
        env={**os.environ},
        capture_output=True,
        text=True,
    )

    from pybFoam import Time, fvMesh, volScalarField

    time = Time(str(CASE.parent), CASE.name)
    mesh = fvMesh(time)
    volScalarField.read_field(mesh, "alpha.water")





.. rst-class:: sphx-glr-script-out

 .. code-block:: none


    <pybFoam.pybFoam_core.volScalarField object at 0x7f46cb50b9d0>



.. GENERATED FROM PYTHON SOURCE LINES 42-46

Define ``Variance``
-------------------
Subclass ``BaseModel``, declare a unique ``type`` literal, implement
``compute(dataset) -> AggregatedDataSet``, register with ``@Node.register()``.

.. GENERATED FROM PYTHON SOURCE LINES 46-73

.. code-block:: Python


    from typing import Literal, Optional

    import numpy as np
    from pydantic import BaseModel

    from pyOFTools.datasets import AggregatedData, AggregatedDataSet, DataSets
    from pyOFTools.node import Node


    @Node.register()
    class Variance(BaseModel):
        """Sample variance of the field. Ignores mask/groups for brevity —
        production nodes should honour both."""

        type: Literal["variance"] = "variance"
        name: Optional[str] = None

        def compute(self, dataset: DataSets) -> AggregatedDataSet:
            values = np.asarray(dataset.field)
            var = float(np.var(values, ddof=0))
            return AggregatedDataSet(
                name=self.name or f"{dataset.name}_variance",
                values=[AggregatedData(value=var)],
            )









.. GENERATED FROM PYTHON SOURCE LINES 74-79

Use it in a pipe
----------------
Once registered, ``Variance`` behaves exactly like a built-in node. The
``|`` operator, ``WorkFlow.then``, and Pydantic validation all work
unchanged.

.. GENERATED FROM PYTHON SOURCE LINES 79-86

.. code-block:: Python


    from pyOFTools.builders import field

    result = (field(mesh, "alpha.water") | Variance()).compute()
    variance_from_node = result.values[0].value
    print(f"Var(alpha.water) via node = {variance_from_node:.6g}")





.. rst-class:: sphx-glr-script-out

 .. code-block:: none

    Var(alpha.water) via node = 0.156096




.. GENERATED FROM PYTHON SOURCE LINES 87-91

Verify against numpy
--------------------
Sanity-check by reading the same field raw and computing the variance in
numpy directly. They should match to floating-point precision.

.. GENERATED FROM PYTHON SOURCE LINES 91-98

.. code-block:: Python


    alpha = volScalarField.from_registry(mesh, "alpha.water")
    raw_values = np.asarray(alpha["internalField"])
    variance_from_numpy = float(np.var(raw_values, ddof=0))
    print(f"Var(alpha.water) via numpy = {variance_from_numpy:.6g}")
    assert np.isclose(variance_from_node, variance_from_numpy), "node and numpy disagree"





.. rst-class:: sphx-glr-script-out

 .. code-block:: none

    Var(alpha.water) via numpy = 0.156096




.. GENERATED FROM PYTHON SOURCE LINES 99-112

Production-grade version
------------------------
The node above is intentionally minimal. Two things a real node should
handle:

1. **Mask and groups** — filter by ``dataset.mask`` and reduce per
   ``dataset.groups`` id. The ``pyOFTools.aggregation`` module has
   MPI-aware helpers (``aggregation.mean``, ``aggregation.sum``) that do
   this for you; see the ``StdDev`` example in
   :doc:`/explanation/custom_nodes`.
2. **Parallel correctness** — the node above computes a local variance
   per rank. For a global variance, compute E[x], E[x^2] with the
   MPI-aware helpers and combine.


.. rst-class:: sphx-glr-timing

   **Total running time of the script:** (0 minutes 10.519 seconds)


.. _sphx_glr_download_auto_how_to_example_custom_aggregator.py:

.. only:: html

  .. container:: sphx-glr-footer sphx-glr-footer-example

    .. container:: sphx-glr-download sphx-glr-download-jupyter

      :download:`Download Jupyter notebook: example_custom_aggregator.ipynb <example_custom_aggregator.ipynb>`

    .. container:: sphx-glr-download sphx-glr-download-python

      :download:`Download Python source code: example_custom_aggregator.py <example_custom_aggregator.py>`

    .. container:: sphx-glr-download sphx-glr-download-zip

      :download:`Download zipped: example_custom_aggregator.zip <example_custom_aggregator.zip>`


.. only:: html

 .. rst-class:: sphx-glr-signature

    `Gallery generated by Sphinx-Gallery <https://sphinx-gallery.github.io>`_