|
| 1 | +# SPDX-License-Identifier: Apache-2.0 |
| 2 | +# SPDX-FileCopyrightText: Copyright 2023-2024 TotalEnergies. |
| 3 | +# SPDX-FileContributor: Raphaël Vinour, Martin Lemay, Romain Baville |
| 4 | +# ruff: noqa: E402 # disable Module level import not at top of file |
| 5 | +import numpy as np |
| 6 | +import numpy.typing as npt |
| 7 | +import logging |
| 8 | + |
| 9 | +from geos.utils.Logger import ( Logger, getLogger ) |
| 10 | +from typing_extensions import Self, Union, Set, List, Dict |
| 11 | + |
| 12 | +from vtkmodules.vtkCommonDataModel import ( |
| 13 | + vtkDataSet, |
| 14 | + vtkMultiBlockDataSet, |
| 15 | +) |
| 16 | + |
| 17 | +from geos.mesh.utils.arrayModifiers import transferAttributeWithElementMap |
| 18 | +from geos.mesh.utils.arrayHelpers import ( computeElementMapping, getAttributeSet, isAttributeGlobal ) |
| 19 | + |
| 20 | +__doc__ = """ |
| 21 | +AttributeMapping is a vtk filter that transfers global attributes from a source mesh to a final mesh with same point/cell coordinates. The final mesh is updated directly, without creation of a copy. |
| 22 | +
|
| 23 | +Input meshes can be vtkDataSet or vtkMultiBlockDataSet. |
| 24 | +
|
| 25 | +.. Warning:: |
| 26 | + For one application of the filter, the attributes to transfer should all be located on the same piece (all on points or all on cells). |
| 27 | +
|
| 28 | +.. Note:: |
| 29 | + For cell, the coordinates of the points in the cell are compared. |
| 30 | + If one of the two meshes is a surface and the other a volume, all the points of the surface must be points of the volume. |
| 31 | +
|
| 32 | +To use a logger handler of yours, set the variable 'speHandler' to True and add it using the member function setLoggerHandler. |
| 33 | +
|
| 34 | +To use the filter: |
| 35 | +
|
| 36 | +.. code-block:: python |
| 37 | +
|
| 38 | + from geos.mesh.processing.AttributeMapping import AttributeMapping |
| 39 | +
|
| 40 | + # Filter inputs. |
| 41 | + meshFrom: Union[ vtkDataSet, vtkMultiBlockDataSet ] |
| 42 | + meshTo: Union[ vtkDataSet, vtkMultiBlockDataSet ] |
| 43 | + attributeNames: Set[ str ] |
| 44 | + # Optional inputs. |
| 45 | + onPoints: bool # defaults to False |
| 46 | + speHandler: bool # defaults to False |
| 47 | +
|
| 48 | + # Instantiate the filter |
| 49 | + filter: AttributeMapping = AttributeMapping( meshFrom, |
| 50 | + meshTo, |
| 51 | + attributeNames, |
| 52 | + onPoints, |
| 53 | + speHandler, |
| 54 | + ) |
| 55 | +
|
| 56 | + # Set the handler of yours (only if speHandler is True). |
| 57 | + yourHandler: logging.Handler |
| 58 | + filter.setLoggerHandler( yourHandler ) |
| 59 | +
|
| 60 | + # Do calculations. |
| 61 | + filter.applyFilter() |
| 62 | +""" |
| 63 | + |
| 64 | +loggerTitle: str = "Attribute Mapping" |
| 65 | + |
| 66 | + |
| 67 | +class AttributeMapping: |
| 68 | + |
| 69 | + def __init__( |
| 70 | + self: Self, |
| 71 | + meshFrom: Union[ vtkDataSet, vtkMultiBlockDataSet ], |
| 72 | + meshTo: Union[ vtkDataSet, vtkMultiBlockDataSet ], |
| 73 | + attributeNames: Set[ str ], |
| 74 | + onPoints: bool = False, |
| 75 | + speHandler: bool = False, |
| 76 | + ) -> None: |
| 77 | + """Transfer global attributes from a source mesh to a final mesh, mapping the piece of the attributes to transfer. |
| 78 | +
|
| 79 | + Args: |
| 80 | + meshFrom (Union[ vtkDataSet, vtkMultiBlockDataSet ]): The source mesh with attributes to transfer. |
| 81 | + meshTo (Union[ vtkDataSet, vtkMultiBlockDataSet ]): The final mesh where to transfer attributes. |
| 82 | + attributeNames (Set[str]): Names of the attributes to transfer. |
| 83 | + onPoints (bool): True if attributes are on points, False if they are on cells. |
| 84 | + Defaults to False. |
| 85 | + speHandler (bool, optional): True to use a specific handler, False to use the internal handler. |
| 86 | + Defaults to False. |
| 87 | + """ |
| 88 | + self.meshFrom: Union[ vtkDataSet, vtkMultiBlockDataSet ] = meshFrom |
| 89 | + self.meshTo: Union[ vtkDataSet, vtkMultiBlockDataSet ] = meshTo |
| 90 | + self.attributeNames: Set[ str ] = attributeNames |
| 91 | + self.onPoints: bool = onPoints |
| 92 | + #TODO/refact (@RomainBaville) make it an enum |
| 93 | + self.piece: str = "points" if self.onPoints else "cells" |
| 94 | + |
| 95 | + # cell map |
| 96 | + self.ElementMap: Dict[ int, npt.NDArray[ np.int64 ] ] = {} |
| 97 | + |
| 98 | + # Logger. |
| 99 | + self.logger: Logger |
| 100 | + if not speHandler: |
| 101 | + self.logger = getLogger( loggerTitle, True ) |
| 102 | + else: |
| 103 | + self.logger = logging.getLogger( loggerTitle ) |
| 104 | + self.logger.setLevel( logging.INFO ) |
| 105 | + |
| 106 | + def setLoggerHandler( self: Self, handler: logging.Handler ) -> None: |
| 107 | + """Set a specific handler for the filter logger. |
| 108 | +
|
| 109 | + In this filter 4 log levels are use, .info, .error, .warning and .critical, be sure to have at least the same 4 levels. |
| 110 | +
|
| 111 | + Args: |
| 112 | + handler (logging.Handler): The handler to add. |
| 113 | + """ |
| 114 | + if not self.logger.hasHandlers(): |
| 115 | + self.logger.addHandler( handler ) |
| 116 | + else: |
| 117 | + self.logger.warning( |
| 118 | + "The logger already has an handler, to use yours set the argument 'speHandler' to True during the filter initialization." |
| 119 | + ) |
| 120 | + |
| 121 | + def getElementMap( self: Self ) -> Dict[ int, npt.NDArray[ np.int64 ] ]: |
| 122 | + """Getter of the element mapping dictionary. |
| 123 | +
|
| 124 | + If attribute to transfer are on points it will be a pointMap, else it will be a cellMap. |
| 125 | +
|
| 126 | + Returns: |
| 127 | + self.elementMap (Dict[int, npt.NDArray[np.int64]]): The element mapping dictionary. |
| 128 | + """ |
| 129 | + return self.ElementMap |
| 130 | + |
| 131 | + def applyFilter( self: Self ) -> bool: |
| 132 | + """Transfer global attributes from a source mesh to a final mesh, mapping the piece of the attributes to transfer. |
| 133 | +
|
| 134 | + Returns: |
| 135 | + boolean (bool): True if calculation successfully ended, False otherwise. |
| 136 | + """ |
| 137 | + self.logger.info( f"Apply filter { self.logger.name }." ) |
| 138 | + |
| 139 | + if len( self.attributeNames ) == 0: |
| 140 | + self.logger.warning( f"Please enter at least one { self.piece } attribute to transfer." ) |
| 141 | + self.logger.warning( f"The filter { self.logger.name } has not been used." ) |
| 142 | + return False |
| 143 | + |
| 144 | + attributesInMeshFrom: Set[ str ] = getAttributeSet( self.meshFrom, self.onPoints ) |
| 145 | + wrongAttributeNames: Set[ str ] = self.attributeNames.difference( attributesInMeshFrom ) |
| 146 | + if len( wrongAttributeNames ) > 0: |
| 147 | + self.logger.error( |
| 148 | + f"The { self.piece } attributes { wrongAttributeNames } are not present in the source mesh." ) |
| 149 | + self.logger.error( f"The filter { self.logger.name } failed." ) |
| 150 | + return False |
| 151 | + |
| 152 | + attributesInMeshTo: Set[ str ] = getAttributeSet( self.meshTo, self.onPoints ) |
| 153 | + attributesAlreadyInMeshTo: Set[ str ] = self.attributeNames.intersection( attributesInMeshTo ) |
| 154 | + if len( attributesAlreadyInMeshTo ) > 0: |
| 155 | + self.logger.error( |
| 156 | + f"The { self.piece } attributes { attributesAlreadyInMeshTo } are already present in the final mesh." ) |
| 157 | + self.logger.error( f"The filter { self.logger.name } failed." ) |
| 158 | + return False |
| 159 | + |
| 160 | + if isinstance( self.meshFrom, vtkMultiBlockDataSet ): |
| 161 | + partialAttributes: List[ str ] = [] |
| 162 | + for attributeName in self.attributeNames: |
| 163 | + if not isAttributeGlobal( self.meshFrom, attributeName, self.onPoints ): |
| 164 | + partialAttributes.append( attributeName ) |
| 165 | + |
| 166 | + if len( partialAttributes ) > 0: |
| 167 | + self.logger.error( |
| 168 | + f"All { self.piece } attributes to transfer must be global, { partialAttributes } are partials." ) |
| 169 | + self.logger.error( f"The filter { self.logger.name } failed." ) |
| 170 | + |
| 171 | + self.ElementMap = computeElementMapping( self.meshFrom, self.meshTo, self.onPoints ) |
| 172 | + sharedElement: bool = False |
| 173 | + for key in self.ElementMap: |
| 174 | + if np.any( self.ElementMap[ key ] > -1 ): |
| 175 | + sharedElement = True |
| 176 | + |
| 177 | + if not sharedElement: |
| 178 | + self.logger.warning( f"The two meshes do not have any shared { self.piece }." ) |
| 179 | + self.logger.warning( f"The filter { self.logger.name } has not been used." ) |
| 180 | + return False |
| 181 | + |
| 182 | + for attributeName in self.attributeNames: |
| 183 | + if not transferAttributeWithElementMap( self.meshFrom, self.meshTo, self.ElementMap, attributeName, |
| 184 | + self.onPoints, self.logger ): |
| 185 | + self.logger.error( f"The attribute { attributeName } has not been mapped." ) |
| 186 | + self.logger.error( f"The filter { self.logger.name } failed." ) |
| 187 | + return False |
| 188 | + |
| 189 | + # Log the output message. |
| 190 | + self._logOutputMessage() |
| 191 | + |
| 192 | + return True |
| 193 | + |
| 194 | + def _logOutputMessage( self: Self ) -> None: |
| 195 | + """Create and log result messages of the filter.""" |
| 196 | + self.logger.info( f"The filter { self.logger.name } succeeded." ) |
| 197 | + self.logger.info( |
| 198 | + f"The { self.piece } attributes { self.attributeNames } have been transferred from the source mesh to the final mesh with the { self.piece } mapping." |
| 199 | + ) |
0 commit comments