feat: add FillPartialArrays vtk filter and paraview plugin

geos-mesh/src/geos/mesh/processing/FillPartialArrays.py

@@ -0,0 +1,166 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright 2023-2024 TotalEnergies.
+# SPDX-FileContributor: Romain Baville, Martin Lemay
+
+from typing_extensions import Self
+from typing import Union, Tuple
+from vtkmodules.util.vtkAlgorithm import VTKPythonAlgorithmBase
+
+from geos.utils.Logger import Logger, getLogger
+from geos.mesh.utils.arrayModifiers import fillPartialAttributes
+from geos.mesh.utils.arrayHelpers import (
+    getNumberOfComponents,
+    isAttributeInObject,
+)
+
+from vtkmodules.vtkCommonCore import (
+    vtkInformation,
+    vtkInformationVector,
+)
+
+from vtkmodules.vtkCommonDataModel import (
+    vtkMultiBlockDataSet, )
+
+import numpy as np
+
+__doc__ = """
+Fill partial arrays of input mesh with values (defaults to nan).
+Several arrays can be filled in one application if the value is the same.
+
+Input and output meshes are vtkMultiBlockDataSet.
+
+To use it:
+
+.. code-block:: python
+
+    from geos.mesh.processing.FillPartialArrays import FillPartialArrays
+
+    # filter inputs
+    input_mesh: vtkMultiBlockDataSet
+    input_attributesNameList: list[str]
+    input_valueToFill: float, optional defaults to nan
+
+    # Instanciate the filter
+    filter: FillPartialArrays = FillPartialArrays()
+    # Set the list of the partial atributes to fill
+    filter._SetAttributesNameList( input_attributesNameList )
+    # Set the value to fill in the partial attributes if not nan
+    filter._SetValueToFill( input_valueToFill )
+    # Set the mesh
+    filter.SetInputDataObject( input_mesh )
+    # Do calculations
+    filter.Update()
+
+    # get output object
+    output: vtkMultiBlockDataSet = filter.GetOutputDataObject( 0 ) )
+"""
+
+
+class FillPartialArrays( VTKPythonAlgorithmBase ):
+
+    def __init__( self: Self ) -> None:
+        """Map the properties of a server mesh to a client mesh."""
+        super().__init__( nInputPorts=1,
+                          nOutputPorts=1,
+                          inputType="vtkMultiBlockDataSet",
+                          outputType="vtkMultiBlockDataSet" )
+
+        # Initialisation of an empty list of the attribute's name
+        self._SetAttributesNameList()
+
+        # Initialisation of the value (nan) to fill in the partial attributes
+        self._SetValueToFill()
+
+        # Logger
+        self.m_logger: Logger = getLogger( "Fill Partial Attributes" )
+
+    def RequestDataObject(
+        self: Self,
+        request: vtkInformation,
+        inInfoVec: list[ vtkInformationVector ],
+        outInfoVec: vtkInformationVector,
+    ) -> int:
+        """Inherited from VTKPythonAlgorithmBase::RequestDataObject.
+
+        Args:
+            request (vtkInformation): Request
+            inInfoVec (list[vtkInformationVector]): Input objects
+            outInfoVec (vtkInformationVector): Output objects
+
+        Returns:
+            int: 1 if calculation successfully ended, 0 otherwise.
+        """
+        inData = self.GetInputData( inInfoVec, 0, 0 )
+        outData = self.GetOutputData( outInfoVec, 0 )
+        assert inData is not None
+        if outData is None or ( not outData.IsA( inData.GetClassName() ) ):
+            outData = inData.NewInstance()
+            outInfoVec.GetInformationObject( 0 ).Set( outData.DATA_OBJECT(), outData )
+        return super().RequestDataObject( request, inInfoVec, outInfoVec )  # type: ignore[no-any-return]
+
+    def RequestData(
+        self: Self,
+        request: vtkInformation,  # noqa: F841
+        inInfoVec: list[ vtkInformationVector ],
+        outInfoVec: vtkInformationVector,
+    ) -> int:
+        """Inherited from VTKPythonAlgorithmBase::RequestData.
+
+        Args:
+            request (vtkInformation): Request
+            inInfoVec (list[vtkInformationVector]): Input objects
+            outInfoVec (vtkInformationVector): Output objects
+
+        Returns:
+            int: 1 if calculation successfully ended, 0 otherwise.
+        """
+        self.m_logger.info( f"Apply filter {__name__}" )
+        try:
+            inputMesh: vtkMultiBlockDataSet = self.GetInputData( inInfoVec, 0, 0 )
+            outData: vtkMultiBlockDataSet = self.GetOutputData( outInfoVec, 0 )
+
+            assert inputMesh is not None, "Input mesh is null."
+            assert outData is not None, "Output pipeline is null."
+
+            outData.ShallowCopy( inputMesh )
+            for attributeName in self._attributesNameList:
+                # cell and point arrays
+                for onPoints in ( False, True ):
+                    if isAttributeInObject( outData, attributeName, onPoints ):
+                        nbComponents = getNumberOfComponents( outData, attributeName, onPoints )
+                        fillPartialAttributes( outData, attributeName, nbComponents, onPoints, self._valueToFill )
+            outData.Modified()
+
+            mess: str = "Fill Partial arrays were successfully completed. " + str(
+                self._attributesNameList ) + " filled with value " + str( self._valueToFill )
+            self.m_logger.info( mess )
+        except AssertionError as e:
+            mess1: str = "Partial arrays filling failed due to:"
+            self.m_logger.error( mess1 )
+            self.m_logger.error( e, exc_info=True )
+            return 0
+        except Exception as e:
+            mess0: str = "Partial arrays filling failed due to:"
+            self.m_logger.critical( mess0 )
+            self.m_logger.critical( e, exc_info=True )
+            return 0
+
+        return 1
+
+    def _SetAttributesNameList( self: Self, attributesNameList: Union[ list[ str ], Tuple ] = () ) -> None:
+        """Set the list of the partial attributes to fill.
+
+        Args:
+            attributesNameList (Union[list[str], Tuple], optional): List of all the attributes name.
+                Defaults to a empty list
+        """
+        self._attributesNameList: Union[ list[ str ], Tuple ] = attributesNameList
+
+    def _SetValueToFill( self: Self, valueToFill: float = np.nan ) -> None:
+        """Set the value to fill in the partial attribute.
+
+        Args:
+            valueToFill (float, optional): The filling value.
+                Defaults to nan.
+        """
+        self._valueToFill: float = valueToFill

geos-mesh/src/geos/mesh/utils/arrayModifiers.py

@@ -39,56 +39,54 @@
 """
 
 
-def fillPartialAttributes(
-    multiBlockMesh: Union[ vtkMultiBlockDataSet, vtkCompositeDataSet, vtkDataObject ],
-    attributeName: str,
-    nbComponents: int,
-    onPoints: bool = False,
-) -> bool:
-    """Fill input partial attribute of multiBlockMesh with nan values.
+def fillPartialAttributes( multiBlockMesh: Union[ vtkMultiBlockDataSet, vtkCompositeDataSet, vtkDataObject ],
+                           attributeName: str,
+                           nbComponents: int,
+                           onPoints: bool = False,
+                           value: float = np.nan ) -> bool:
+    """Fill input partial attribute of multiBlockMesh with values (defaults to nan).
 
     Args:
         multiBlockMesh (vtkMultiBlockDataSet | vtkCompositeDataSet | vtkDataObject): multiBlock
-            mesh where to fill the attribute
-        attributeName (str): attribute name
-        nbComponents (int): number of components
-        onPoints (bool, optional): Attribute is on Points (False) or
-            on Cells.
-
+            mesh where to fill the attribute.
+        attributeName (str): Attribute name.
+        nbComponents (int): Number of components.
+        onPoints (bool, optional): Attribute is on Points (True) or on Cells (False).
             Defaults to False.
+        value (float, optional): The filling value.
+            Defaults to nan.
 
     Returns:
-        bool: True if calculation successfully ended, False otherwise
+        bool: True if calculation successfully ended, False otherwise.
     """
     componentNames: tuple[ str, ...] = ()
     if nbComponents > 1:
         componentNames = getComponentNames( multiBlockMesh, attributeName, onPoints )
-    values: list[ float ] = [ np.nan for _ in range( nbComponents ) ]
+    values: list[ float ] = [ value for _ in range( nbComponents ) ]
     createConstantAttribute( multiBlockMesh, values, attributeName, componentNames, onPoints )
     multiBlockMesh.Modified()
     return True
 
 
-def fillAllPartialAttributes(
-    multiBlockMesh: Union[ vtkMultiBlockDataSet, vtkCompositeDataSet, vtkDataObject ],
-    onPoints: bool = False,
-) -> bool:
-    """Fill all the partial attributes of multiBlockMesh with nan values.
+def fillAllPartialAttributes( multiBlockMesh: Union[ vtkMultiBlockDataSet, vtkCompositeDataSet, vtkDataObject ],
+                              onPoints: bool = False,
+                              value: float = np.nan ) -> bool:
+    """Fill all the partial attributes of multiBlockMesh with values (defaults to nan).
 
     Args:
         multiBlockMesh (vtkMultiBlockDataSet | vtkCompositeDataSet | vtkDataObject):
             multiBlockMesh where to fill the attribute
-        onPoints (bool, optional): Attribute is on Points (False) or
-            on Cells.
-
+        onPoints (bool, optional): Attribute is on Points (True) or on Cells (False).
             Defaults to False.
+        value (float, optional): The filling value.
+            Defaults to nan.
 
     Returns:
         bool: True if calculation successfully ended, False otherwise
     """
     attributes: dict[ str, int ] = getAttributesWithNumberOfComponents( multiBlockMesh, onPoints )
     for attributeName, nbComponents in attributes.items():
-        fillPartialAttributes( multiBlockMesh, attributeName, nbComponents, onPoints )
+        fillPartialAttributes( multiBlockMesh, attributeName, nbComponents, onPoints, value )
     multiBlockMesh.Modified()
     return True
 

geos-mesh/tests/conftest.py

@@ -51,4 +51,4 @@ def _get_dataset( datasetType: str ):
 
         return reader.GetOutput()
 
-    return _get_dataset
+    return _get_dataset

geos-mesh/tests/test_FillPartialArrays.py

@@ -0,0 +1,75 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright 2023-2024 TotalEnergies.
+# SPDX-FileContributor: Romain Baville
+# SPDX-License-Identifier: Apache 2.0
+# ruff: noqa: E402 # disable Module level import not at top of file
+# mypy: disable-error-code="operator"
+import pytest
+from typing import Union, Tuple, cast
+
+import numpy as np
+import numpy.typing as npt
+
+import vtkmodules.util.numpy_support as vnp
+from vtkmodules.vtkCommonDataModel import ( vtkDataSet, vtkMultiBlockDataSet, vtkPointData, vtkCellData )
+
+from geos.mesh.processing.FillPartialArrays import FillPartialArrays
+
+
+@pytest.mark.parametrize( "onpoints, attributesList, value_test", [
+    ( False, ( ( 0, "PORO", 1 ), ), np.nan ),
+    ( True, ( ( 0, "PointAttribute", 3 ), ( 1, "collocated_nodes", 2 ) ), 2. ),
+    ( False, ( ( 0, "CELL_MARKERS", 1 ), ( 0, "CellAttribute", 3 ), ( 0, "FAULT", 1 ), ( 0, "PERM", 3 ),
+               ( 0, "PORO", 1 ) ), 2. ),
+    ( False, ( ( 0, "PORO", 1 ), ), 2.0 ),
+    ( True, ( ( 0, "PointAttribute", 3 ), ( 1, "collocated_nodes", 2 ) ), np.nan ),
+    ( False, ( ( 0, "CELL_MARKERS", 1 ), ( 0, "CellAttribute", 3 ), ( 0, "FAULT", 1 ), ( 0, "PERM", 3 ),
+               ( 0, "PORO", 1 ) ), np.nan ),
+] )
+def test_FillPartialArrays(
+    dataSetTest: vtkMultiBlockDataSet,
+    onpoints: bool,
+    attributesList: Tuple[ Tuple[ int, str, int ], ...],
+    value_test: float,
+) -> None:
+    """Test FillPartialArrays vtk filter."""
+    vtkMultiBlockDataSetTestRef: vtkMultiBlockDataSet = dataSetTest( "multiblock" )
+    vtkMultiBlockDataSetTest: vtkMultiBlockDataSet = dataSetTest( "multiblock" )
+    attributesNameList: list[ str ] = [ attributesList[ i ][ 1 ] for i in range( len( attributesList ) ) ]
+
+    filter: FillPartialArrays = FillPartialArrays()
+    filter._SetAttributesNameList( attributesNameList )
+    filter._SetValueToFill( value_test )
+    filter.SetInputDataObject( vtkMultiBlockDataSetTest )
+    filter.Update()
+
+    nbBlock: int = vtkMultiBlockDataSetTestRef.GetNumberOfBlocks()
+    for block_id in range( nbBlock ):
+        datasetRef: vtkDataSet = cast( vtkDataSet, vtkMultiBlockDataSetTestRef.GetBlock( block_id ) )
+        dataset: vtkDataSet = cast( vtkDataSet, filter.GetOutputDataObject( 0 ).GetBlock( block_id ) )
+        expected_array: npt.NDArray[ np.float64 ]
+        array: npt.NDArray[ np.float64 ]
+        dataRef: Union[ vtkPointData, vtkCellData ]
+        data: Union[ vtkPointData, vtkCellData ]
+        nbElements: list[ int ]
+        if onpoints:
+            dataRef = datasetRef.GetPointData()
+            data = dataset.GetPointData()
+            nbElements = [ 212, 4092 ]
+        else:
+            dataRef = datasetRef.GetCellData()
+            data = dataset.GetCellData()
+            nbElements = [ 156, 1740 ]
+
+        for inBlock, attribute, nbComponents in attributesList:
+            array = vnp.vtk_to_numpy( data.GetArray( attribute ) )
+            if block_id == inBlock:
+                expected_array = vnp.vtk_to_numpy( dataRef.GetArray( attribute ) )
+                assert ( array == expected_array ).all()
+            else:
+                expected_array = np.array( [ [ value_test for i in range( nbComponents ) ]
+                                             for _ in range( nbElements[ inBlock ] ) ] )
+                if np.isnan( value_test ):
+                    assert np.all( np.isnan( array ) == np.isnan( expected_array ) )
+                else:
+                    assert ( array == expected_array ).all()