diff --git a/docs/geos_pv_docs/processing.rst b/docs/geos_pv_docs/processing.rst index 36800919..6b7c00ff 100644 --- a/docs/geos_pv_docs/processing.rst +++ b/docs/geos_pv_docs/processing.rst @@ -53,36 +53,17 @@ PVGeosBlockExtractAndMerge plugin .. automodule:: geos.pv.plugins.PVGeosBlockExtractAndMerge - Geomechanics workflows ++++++++++++++++++++++++ -PVGeomechanicsWorkflowVolume plugin ---------------------------------------------- - -.. automodule:: geos.pv.plugins.PVGeomechanicsWorkflowVolume - - -PVGeomechanicsWorkflowVolumeSurface plugin ----------------------------------------------------- - -.. automodule:: geos.pv.plugins.PVGeomechanicsWorkflowVolumeSurface - - -PVGeomechanicsWorkflowVolumeSurfaceWell plugin --------------------------------------------------------- - -.. automodule:: geos.pv.plugins.PVGeomechanicsWorkflowVolumeSurfaceWell - - -PVGeomechanicsWorkflowVolumeWell plugin -------------------------------------------------- +PVGeomechanicsWorkflow plugin +------------------------------ -.. automodule:: geos.pv.plugins.PVGeomechanicsWorkflowVolumeWell +.. automodule:: geos.pv.plugins.PVGeomechanicsWorkflow PVGeomechanicsCalculator plugin --------------------------------------- -.. automodule:: geos.pv.plugins.PVGeomechanicsCalculator \ No newline at end of file +.. automodule:: geos.pv.plugins.PVGeomechanicsCalculator diff --git a/geos-mesh/src/geos/mesh/utils/genericHelpers.py b/geos-mesh/src/geos/mesh/utils/genericHelpers.py index 0c3318ea..de11a402 100644 --- a/geos-mesh/src/geos/mesh/utils/genericHelpers.py +++ b/geos-mesh/src/geos/mesh/utils/genericHelpers.py @@ -62,16 +62,17 @@ def triangulateMesh( Returns: vtkUnstructuredGrid: Triangulated mesh """ + vtkErrorLogger: Logger if logger is None: - logger = getLogger( "Triangulate", True ) - # Creation of a child logger to deal with VTKErrors without polluting parent logger - vtkErrorLogger: Logger = getLogger( f"{logger.name}.vtkErrorLogger", True ) - vtkErrorLogger.propagate = False + vtkErrorLogger = getLogger( "Triangulate Mesh vtkError Logger", True ) + else: + vtkErrorLogger = logging.getLogger( f"{ logger.name } vtkError Logger" ) + vtkErrorLogger.setLevel( logging.INFO ) + vtkErrorLogger.addHandler( logger.handlers[ 0 ] ) + vtkErrorLogger.propagate = False vtkLogger.SetStderrVerbosity( vtkLogger.VERBOSITY_ERROR ) - vtkErrorLogger.addFilter( RegexExceptionFilter() ) # will raise VTKError if captured VTK Error - vtkErrorLogger.setLevel( logging.DEBUG ) with VTKCaptureLog() as capturedLog: triangulateMeshFilter: vtkDataSetTriangleFilter = vtkDataSetTriangleFilter() @@ -81,7 +82,8 @@ def triangulateMesh( capturedLog.seek( 0 ) captured = capturedLog.read().decode() - vtkErrorLogger.debug( captured.strip() ) + if captured != "": + vtkErrorLogger.error( captured.strip() ) triangulatedMesh: vtkUnstructuredGrid = triangulateMeshFilter.GetOutput() @@ -107,16 +109,17 @@ def convertUnstructuredGridToPolyData( Returns: vtkPolyData: Extracted surface """ + vtkErrorLogger: Logger if logger is None: - logger = getLogger( "ConvertVtkUnstructuredGridToVtkPolyData.", True ) - # Creation of a child logger to deal with VTKErrors without polluting parent logger - vtkErrorLogger: Logger = getLogger( f"{logger.name}.vtkErrorLogger", True ) - vtkErrorLogger.propagate = False + vtkErrorLogger = getLogger( "Convert vtkUnstructuredGrid To vtkPolyData vtkError Logger", True ) + else: + vtkErrorLogger = logging.getLogger( f"{ logger.name } vtkError Logger" ) + vtkErrorLogger.setLevel( logging.INFO ) + vtkErrorLogger.addHandler( logger.handlers[ 0 ] ) + vtkErrorLogger.propagate = False vtkLogger.SetStderrVerbosity( vtkLogger.VERBOSITY_ERROR ) - vtkErrorLogger.addFilter( RegexExceptionFilter() ) # will raise VTKError if captured VTK Error - vtkErrorLogger.setLevel( logging.DEBUG ) with VTKCaptureLog() as capturedLog: extractSurfaceFilter: vtkDataSetSurfaceFilter = vtkDataSetSurfaceFilter() @@ -131,7 +134,8 @@ def convertUnstructuredGridToPolyData( capturedLog.seek( 0 ) captured = capturedLog.read().decode() - vtkErrorLogger.debug( captured.strip() ) + if captured != "": + vtkErrorLogger.error( captured.strip() ) extractedSurface: vtkPolyData = extractSurfaceFilter.GetOutput() @@ -460,8 +464,8 @@ def getTangentsVectors( surface: vtkPolyData ) -> Tuple[ npt.NDArray[ np.float64 try: tangents1 = vtk_to_numpy( vtkTangents ) except AttributeError as err: - print( "No tangential attribute found in the mesh. Use the computeTangents function beforehand." ) - raise VTKError( err ) from err + context: str = f"No tangential attribute found in the mesh. Use the computeTangents function beforehand.\n{ err }" + raise VTKError( context ) from err else: # Compute second tangential component normals: npt.NDArray[ np.float64 ] = getNormalVectors( surface ) @@ -471,22 +475,30 @@ def getTangentsVectors( surface: vtkPolyData ) -> Tuple[ npt.NDArray[ np.float64 return ( tangents1, tangents2 ) -def getLocalBasisVectors( surface: vtkPolyData ) -> npt.NDArray[ np.float64 ]: +def getLocalBasisVectors( + surface: vtkPolyData, + logger: Union[ Logger, None ] = None, +) -> npt.NDArray[ np.float64 ]: """Return the local basis vectors for all cells of the input surface. Args: surface(vtkPolydata): The input surface. + logger (Union[Logger, None], optional): A logger to manage the output messages. + Defaults to None, an internal logger is used. Returns: npt.NDArray[np.float64]: Array with normal, tangential 1 and tangential 2 vectors. """ + if logger is None: + logger = getLogger( "getLocalBasisVectors", True ) + try: normals: npt.NDArray[ np.float64 ] = getNormalVectors( surface ) surfaceWithNormals: vtkPolyData = surface # ValueError raised if no normals found in the mesh except ValueError: # In that case, the normals are computed. - surfaceWithNormals = computeNormals( surface ) + surfaceWithNormals = computeNormals( surface, logger ) normals = getNormalVectors( surfaceWithNormals ) # Tangents require normals to be present in the mesh @@ -495,7 +507,7 @@ def getLocalBasisVectors( surface: vtkPolyData ) -> npt.NDArray[ np.float64 ]: npt.NDArray[ np.float64 ] ] = getTangentsVectors( surfaceWithNormals ) # If no tangents is present in the mesh, they are computed on that surface except VTKError: - surfaceWithTangents: vtkPolyData = computeTangents( surfaceWithNormals ) + surfaceWithTangents: vtkPolyData = computeTangents( surfaceWithNormals, logger ) tangents = getTangentsVectors( surfaceWithTangents ) return np.array( ( normals, *tangents ) ) @@ -515,16 +527,17 @@ def computeNormals( Returns: vtkPolyData: The surface with normal attribute. """ + vtkErrorLogger: Logger if logger is None: - logger = getLogger( "computeSurfaceNormals" ) - # Creation of a child logger to deal with VTKErrors without polluting parent logger - vtkErrorLogger: Logger = getLogger( f"{logger.name}.vtkErrorLogger" ) - vtkErrorLogger.propagate = False + vtkErrorLogger = getLogger( "Compute Surface Normals vtkError Logger", True ) + else: + vtkErrorLogger = logging.getLogger( f"{ logger.name } vtkError Logger" ) + vtkErrorLogger.setLevel( logging.INFO ) + vtkErrorLogger.addHandler( logger.handlers[ 0 ] ) + vtkErrorLogger.propagate = False vtkLogger.SetStderrVerbosity( vtkLogger.VERBOSITY_ERROR ) - vtkErrorLogger.addFilter( RegexExceptionFilter() ) # will raise VTKError if captured VTK Error - vtkErrorLogger.setLevel( logging.DEBUG ) with VTKCaptureLog() as capturedLog: normalFilter: vtkPolyDataNormals = vtkPolyDataNormals() @@ -536,7 +549,8 @@ def computeNormals( capturedLog.seek( 0 ) captured = capturedLog.read().decode() - vtkErrorLogger.debug( captured.strip() ) + if captured != "": + vtkErrorLogger.error( captured.strip() ) outputSurface = normalFilter.GetOutput() @@ -562,22 +576,23 @@ def computeTangents( Returns: vtkPolyData: The surface with tangent attribute """ - # need to compute texture coordinates required for tangent calculation - surface1: vtkPolyData = computeSurfaceTextureCoordinates( triangulatedSurface ) + # Need to compute texture coordinates required for tangent calculation + surface1: vtkPolyData = computeSurfaceTextureCoordinates( triangulatedSurface, logger ) # TODO: triangulate the surface before computation of the tangents if needed. - # compute tangents + # Compute tangents + vtkErrorLogger: Logger if logger is None: - logger = getLogger( "computeSurfaceTangents" ) - # Creation of a child logger to deal with VTKErrors without polluting parent logger - vtkErrorLogger: Logger = getLogger( f"{logger.name}.vtkErrorLogger" ) - vtkErrorLogger.propagate = False + vtkErrorLogger = getLogger( "Compute Surface Tangents vtkError Logger", True ) + else: + vtkErrorLogger = logging.getLogger( f"{ logger.name } vtkError Logger" ) + vtkErrorLogger.setLevel( logging.INFO ) + vtkErrorLogger.addHandler( logger.handlers[ 0 ] ) + vtkErrorLogger.propagate = False vtkLogger.SetStderrVerbosity( vtkLogger.VERBOSITY_ERROR ) - vtkErrorLogger.addFilter( RegexExceptionFilter() ) # will raise VTKError if captured VTK Error - vtkErrorLogger.setLevel( logging.DEBUG ) with VTKCaptureLog() as capturedLog: @@ -591,7 +606,8 @@ def computeTangents( capturedLog.seek( 0 ) captured = capturedLog.read().decode() - vtkErrorLogger.debug( captured.strip() ) + if captured != "": + vtkErrorLogger.error( captured.strip() ) if surfaceOut is None: raise VTKError( "Something went wrong in VTK calculation." ) @@ -623,16 +639,17 @@ def computeSurfaceTextureCoordinates( vtkPolyData: The input surface with generated texture map. """ # Need to compute texture coordinates required for tangent calculation + vtkErrorLogger: Logger if logger is None: - logger = getLogger( "computeSurfaceTextureCoordinates" ) - # Creation of a child logger to deal with VTKErrors without polluting parent logger - vtkErrorLogger: Logger = getLogger( f"{logger.name}.vtkErrorLogger" ) - vtkErrorLogger.propagate = False + vtkErrorLogger = getLogger( "Compute Surface Texture Coordinates vtkError Logger", True ) + else: + vtkErrorLogger = logging.getLogger( f"{ logger.name } vtkError Logger" ) + vtkErrorLogger.setLevel( logging.INFO ) + vtkErrorLogger.addHandler( logger.handlers[ 0 ] ) + vtkErrorLogger.propagate = False vtkLogger.SetStderrVerbosity( vtkLogger.VERBOSITY_ERROR ) - vtkErrorLogger.addFilter( RegexExceptionFilter() ) # will raise VTKError if captured VTK Error - vtkErrorLogger.setLevel( logging.DEBUG ) with VTKCaptureLog() as capturedLog: @@ -644,6 +661,7 @@ def computeSurfaceTextureCoordinates( capturedLog.seek( 0 ) captured = capturedLog.read().decode() - vtkErrorLogger.debug( captured.strip() ) + if captured != "": + vtkErrorLogger.error( captured.strip() ) return textureFilter.GetOutput() diff --git a/geos-processing/src/geos/processing/post_processing/GeomechanicsCalculator.py b/geos-processing/src/geos/processing/post_processing/GeomechanicsCalculator.py index ffd8dd4a..05340469 100644 --- a/geos-processing/src/geos/processing/post_processing/GeomechanicsCalculator.py +++ b/geos-processing/src/geos/processing/post_processing/GeomechanicsCalculator.py @@ -85,6 +85,7 @@ mesh: vtkUnstructuredGrid computeAdvancedProperties: bool # optional, defaults to False speHandler: bool # optional, defaults to False + loggerName: str # Defaults to "Geomechanics Calculator" # Instantiate the filter geomechanicsCalculatorFilter: GeomechanicsCalculator = GeomechanicsCalculator( mesh, computeAdvancedProperties, speHandler ) @@ -693,7 +694,7 @@ def __init__( computeAdvancedProperties (bool, optional): True to compute advanced geomechanics properties, False otherwise. Defaults to False. loggerName (str, optional): Name of the filter logger. - Defaults to "Geomechanics Calculator" + Defaults to "Geomechanics Calculator". speHandler (bool, optional): True to use a specific handler, False to use the internal handler. Defaults to False. """ diff --git a/geos-processing/src/geos/processing/post_processing/GeosBlockExtractor.py b/geos-processing/src/geos/processing/post_processing/GeosBlockExtractor.py index a4ff7cfd..985959aa 100644 --- a/geos-processing/src/geos/processing/post_processing/GeosBlockExtractor.py +++ b/geos-processing/src/geos/processing/post_processing/GeosBlockExtractor.py @@ -38,6 +38,7 @@ extractFault: bool # Defaults to False extractWell: bool # Defaults to False speHandler: bool # Defaults to False + loggerName: str # Defaults to "Geos Block Extractor" # Instantiate the filter geosBlockExtractor: GeosBlockExtractor = GeosBlockExtractor( geosMesh, extractFault, extractWell, speHandler ) @@ -56,8 +57,6 @@ geosDomainExtracted = geosBlockExtractor.extractedGeosDomain.well # For well domain """ -loggerTitle: str = "Geos Block Extractor" - class GeosExtractDomainBlock( vtkExtractBlock ): @@ -149,6 +148,7 @@ def __init__( extractFault: bool = False, extractWell: bool = False, speHandler: bool = False, + loggerName: str = "Geos Block Extractor", ) -> None: """Blocks from the ElementRegions from a GEOS output multiBlockDataset mesh. @@ -160,6 +160,8 @@ def __init__( Defaults to False. speHandler (bool, optional): True to use a specific handler, False to use the internal handler. Defaults to False. + loggerName (str, optional): Name of the filter logger. + Defaults to "Geos Block Extractor". """ self.geosMesh: vtkMultiBlockDataSet = geosMesh self.extractedGeosDomain = self.ExtractedGeosDomain() @@ -173,9 +175,9 @@ def __init__( # Logger. self.logger: Logger if not speHandler: - self.logger = getLogger( loggerTitle, True ) + self.logger = getLogger( loggerName, True ) else: - self.logger = logging.getLogger( loggerTitle ) + self.logger = logging.getLogger( loggerName ) self.logger.setLevel( logging.INFO ) def setLoggerHandler( self: Self, handler: logging.Handler ) -> None: @@ -201,15 +203,18 @@ def applyFilter( self: Self ) -> None: extractGeosDomain: GeosExtractDomainBlock = GeosExtractDomainBlock() extractGeosDomain.SetInputData( self.geosMesh ) + domainNames: list = [] for domain in self.domainToExtract: extractGeosDomain.RemoveAllIndices() extractGeosDomain.AddGeosDomainName( domain ) extractGeosDomain.Update() self.extractedGeosDomain.setExtractedDomain( domain, extractGeosDomain.GetOutput() ) + domainNames.append( domain.value ) + + self.logger.info( f"The GEOS domain { domainNames } have been extracted." ) + self.logger.info( f"The filter { self.logger.name } succeeded." ) - self.logger.info( "The filter succeeded." ) + except ( ValueError, TypeError ) as e: + self.logger.error( f"The filter { self.logger.name } failed.\n{ e }." ) - except ValueError as ve: - self.logger.error( f"The filter failed.\n{ ve }." ) - except TypeError as te: - self.logger.error( f"The filter failed.\n{ te }." ) + return diff --git a/geos-processing/src/geos/processing/post_processing/GeosBlockMerge.py b/geos-processing/src/geos/processing/post_processing/GeosBlockMerge.py index 09efe6a6..fac8c108 100644 --- a/geos-processing/src/geos/processing/post_processing/GeosBlockMerge.py +++ b/geos-processing/src/geos/processing/post_processing/GeosBlockMerge.py @@ -44,9 +44,10 @@ # Optional inputs. convertFaultToSurface: bool # Defaults to False speHandler: bool # Defaults to False + loggerName: str # Defaults to "GEOS Block Merge" # Instantiate the filter - mergeBlockFilter: GeosBlockMerge = GeosBlockMerge( inputMesh, convertFaultToSurface, speHandler ) + mergeBlockFilter: GeosBlockMerge = GeosBlockMerge( inputMesh, convertFaultToSurface, speHandler, loggerName ) # Set the handler of yours (only if speHandler is True). yourHandler: logging.Handler @@ -59,8 +60,6 @@ outputMesh: vtkMultiBlockDataSet = mergeBlockFilter.getOutput() """ -loggerTitle: str = "GEOS Block Merge" - class GeosBlockMerge(): @@ -69,6 +68,7 @@ def __init__( inputMesh: vtkMultiBlockDataSet, convertFaultToSurface: bool = False, speHandler: bool = False, + loggerName: str = "GEOS Block Merge", ) -> None: """VTK Filter that merges ranks of GEOS output mesh. @@ -83,10 +83,13 @@ def __init__( Defaults to False. speHandler (bool, optional): True to use a specific handler, False to use the internal handler. Defaults to False. + loggerName (str, optional): Name of the filter logger. + Defaults to "GEOS Block Merge". """ self.inputMesh: vtkMultiBlockDataSet = inputMesh self.convertFaultToSurface: bool = convertFaultToSurface + self.handler: None | logging.Handler = None self.outputMesh: vtkMultiBlockDataSet = vtkMultiBlockDataSet() self.phaseNameDict: dict[ str, set[ str ] ] = { @@ -94,13 +97,14 @@ def __init__( PhaseTypeEnum.FLUID.type: set(), } - # Logger. + # Logger self.logger: Logger if not speHandler: - self.logger = getLogger( loggerTitle, True ) + self.logger = getLogger( loggerName, True ) else: - self.logger = logging.getLogger( loggerTitle ) + self.logger = logging.getLogger( loggerName ) self.logger.setLevel( logging.INFO ) + self.logger.propagate = False def setLoggerHandler( self: Self, handler: logging.Handler ) -> None: """Set a specific handler for the filter logger. @@ -111,6 +115,7 @@ def setLoggerHandler( self: Self, handler: logging.Handler ) -> None: handler (logging.Handler): The handler to add. """ if not self.logger.hasHandlers(): + self.handler = handler self.logger.addHandler( handler ) else: self.logger.warning( @@ -161,8 +166,8 @@ def applyFilter( self: Self ) -> None: # Convert the volume mesh to a surface mesh if self.convertFaultToSurface: if not isTriangulate( volumeMesh ): - volumeMesh.ShallowCopy( triangulateMesh( volumeMesh ) ) - surfaceMesh: vtkPolyData = convertUnstructuredGridToPolyData( volumeMesh ) + volumeMesh.ShallowCopy( triangulateMesh( volumeMesh, self.logger ) ) + surfaceMesh: vtkPolyData = convertUnstructuredGridToPolyData( volumeMesh, self.logger ) surfaceMesh.ShallowCopy( computeNormals( surfaceMesh, logger=self.logger ) ) surfaceMesh.ShallowCopy( computeTangents( surfaceMesh, logger=self.logger ) ) # Add the merged block to the output mesh @@ -170,9 +175,9 @@ def applyFilter( self: Self ) -> None: else: self.outputMesh.SetBlock( newIndex, volumeMesh ) - self.logger.info( "The filter succeeded." ) + self.logger.info( f"The filter { self.logger.name } succeeded." ) except ( ValueError, TypeError, RuntimeError, AssertionError, VTKError ) as e: - self.logger.critical( f"The filter failed.\n{ e }" ) + self.logger.error( f"The filter { self.logger.name } failed.\n{ e }" ) return @@ -197,6 +202,8 @@ def renameAttributes( else: renameAttribute( mesh, attributeName, newName, False ) + return + def computePhaseNames( self: Self ) -> None: """Get the names of the phases in the mesh from Cell attributes.""" # All the phase attributes are on cells diff --git a/geos-processing/src/geos/processing/post_processing/SurfaceGeomechanics.py b/geos-processing/src/geos/processing/post_processing/SurfaceGeomechanics.py index 90320e82..62fcd013 100644 --- a/geos-processing/src/geos/processing/post_processing/SurfaceGeomechanics.py +++ b/geos-processing/src/geos/processing/post_processing/SurfaceGeomechanics.py @@ -351,7 +351,7 @@ def __computeXYZCoordinates( attrXYZ: npt.NDArray[ np.float64 ] = np.full_like( attrArray, np.nan ) # Get all local basis vectors - localBasis: npt.NDArray[ np.float64 ] = getLocalBasisVectors( self.outputMesh ) + localBasis: npt.NDArray[ np.float64 ] = getLocalBasisVectors( self.outputMesh, self.logger ) for i, cellAttribute in enumerate( attrArray ): if len( cellAttribute ) not in ( 3, 6, 9 ): diff --git a/geos-pv/src/geos/pv/plugins/PVGeomechanicsCalculator.py b/geos-pv/src/geos/pv/plugins/PVGeomechanicsCalculator.py index e0de4fbe..5d870d75 100644 --- a/geos-pv/src/geos/pv/plugins/PVGeomechanicsCalculator.py +++ b/geos-pv/src/geos/pv/plugins/PVGeomechanicsCalculator.py @@ -3,17 +3,19 @@ # SPDX-FileContributor: Martin Lemay, Romain Baville # ruff: noqa: E402 # disable Module level import not at top of file import sys +import numpy as np + from pathlib import Path from typing_extensions import Self from paraview.util.vtkAlgorithm import ( # type: ignore[import-not-found] - VTKPythonAlgorithmBase, smdomain, smproperty, + VTKPythonAlgorithmBase, smdomain, smproperty ) # source: https://github.com/Kitware/ParaView/blob/master/Wrapping/Python/paraview/util/vtkAlgorithm.py from paraview.detail.loghandler import ( # type: ignore[import-not-found] - VTKHandler, + VTKHandler ) # source: https://github.com/Kitware/ParaView/blob/master/Wrapping/Python/paraview/detail/loghandler.py -from vtkmodules.vtkCommonDataModel import vtkUnstructuredGrid, vtkMultiBlockDataSet +from vtkmodules.vtkCommonDataModel import ( vtkUnstructuredGrid, vtkMultiBlockDataSet ) # update sys.path to load all GEOS Python Package dependencies geos_pv_path: Path = Path( __file__ ).parent.parent.parent.parent.parent @@ -23,7 +25,7 @@ update_paths() from geos.utils.PhysicalConstants import ( - DEFAULT_FRICTION_ANGLE_RAD, + DEFAULT_FRICTION_ANGLE_DEG, DEFAULT_GRAIN_BULK_MODULUS, DEFAULT_ROCK_COHESION, WATER_DENSITY, @@ -91,7 +93,7 @@ def __init__( self: Self ) -> None: self.specificDensity: float = WATER_DENSITY ## For advanced properties self.rockCohesion: float = DEFAULT_ROCK_COHESION - self.frictionAngle: float = DEFAULT_FRICTION_ANGLE_RAD + self.frictionAngle: float = DEFAULT_FRICTION_ANGLE_DEG @smproperty.doublevector( name="GrainBulkModulus", @@ -190,23 +192,23 @@ def setRockCohesion( self: Self, rockCohesion: float ) -> None: @smproperty.doublevector( name="FrictionAngle", - label="Friction Angle (rad)", - default_values=DEFAULT_FRICTION_ANGLE_RAD, + label="Friction Angle (°)", + default_values=DEFAULT_FRICTION_ANGLE_DEG, panel_visibility="default", ) @smdomain.xml( """ Reference friction angle to compute critical pore pressure. - The unit is rad. Default is 10.0 / 180.0 * np.pi rad. + The unit is °. Default is no friction case (i.e., 0.°). """ ) def setFrictionAngle( self: Self, frictionAngle: float ) -> None: """Set friction angle. Args: - frictionAngle (float): Friction angle (rad). + frictionAngle (float): Friction angle (°). """ - self.frictionAngle = frictionAngle + self.frictionAngle = frictionAngle * np.pi / 180 self.Modified() @smproperty.xml( """ diff --git a/geos-pv/src/geos/pv/plugins/PVGeomechanicsWorkflow.py b/geos-pv/src/geos/pv/plugins/PVGeomechanicsWorkflow.py new file mode 100644 index 00000000..992a7302 --- /dev/null +++ b/geos-pv/src/geos/pv/plugins/PVGeomechanicsWorkflow.py @@ -0,0 +1,402 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright 2023-2024 TotalEnergies. +# SPDX-FileContributor: Martin Lemay, Romain Baville +# ruff: noqa: E402 # disable Module level import not at top of file +import sys +import logging + +from pathlib import Path +from typing_extensions import Self + +# update sys.path to load all GEOS Python Package dependencies +geos_pv_path: Path = Path( __file__ ).parent.parent.parent.parent.parent +sys.path.insert( 0, str( geos_pv_path / "src" ) ) +from geos.pv.utils.config import update_paths + +update_paths() + +from geos.utils.PhysicalConstants import ( + DEFAULT_FRICTION_ANGLE_DEG, + DEFAULT_GRAIN_BULK_MODULUS, + DEFAULT_ROCK_COHESION, + WATER_DENSITY, +) + +from geos.pv.plugins.PVGeosBlockExtractAndMerge import PVGeosBlockExtractAndMerge +from geos.pv.plugins.PVGeomechanicsCalculator import PVGeomechanicsCalculator +from geos.pv.plugins.PVSurfaceGeomechanics import PVSurfaceGeomechanics + +from vtkmodules.vtkCommonCore import ( vtkInformation, vtkInformationVector ) +from vtkmodules.vtkCommonDataModel import vtkMultiBlockDataSet + +from paraview.util.vtkAlgorithm import ( # type: ignore[import-not-found] + VTKPythonAlgorithmBase, smdomain, smproperty, smproxy ) +from paraview.detail.loghandler import ( # type: ignore[import-not-found] + VTKHandler ) + +__doc__ = """ +PVGeomechanicsWorkflow is a Paraview plugin that executes multiple plugins: + 1. PVGeosBlockExtractAndMerge + 2. PVGeomechanicsCalculator + 3. PVSurfaceGeomechanics (if the input mesh contains faults) + +PVGeosBlockExtractAndMerge is a Paraview plugin processing the input mesh at the current time in two steps: + 1. Extraction of domains (volume, fault and well) from a GEOS output multiBlockDataSet mesh + 2. Actions on each region of a GEOS output domain (volume, fault, wells) to: + * Merge Ranks + * Identify "Fluids" and "Rock" phases + * Rename "Rock" attributes depending on the phase they refer to for more clarity + * Convert volume meshes to surface if needed + * Copy "geomechanics" attributes from the initial timestep to the current one if they exist + +PVGeomechanicsCalculator is a paraview plugin that allows to compute basic and advanced geomechanics properties from existing ones in the mesh. This is donne on each block of the volume mesh. + +The basic geomechanics properties computed on the mesh are: + - The elastic moduli not present on the mesh + - Biot coefficient + - Compressibility, oedometric compressibility and real compressibility coefficient + - Specific gravity + - Real effective stress ratio + - Total initial stress, total current stress and total stress ratio + - Elastic stain + - Real reservoir stress path and reservoir stress path in oedometric condition + +The advanced geomechanics properties computed on the mesh are: + - Fracture index and threshold + - Critical pore pressure and pressure index + +PVSurfaceGeomechanics is a Paraview plugin that allows to compute additional geomechanical attributes from the input surfaces, such as shear capacity utilization (SCU). This is donne on each block of the fault mesh. + +This filter results in 3 output pipelines with the vtkMultiBlockDataSet: + - "Volume" contains the volume domain + - "Fault" contains the fault domain if it exist + - "Well" contains the well domain if it exist + +Input and output meshes are vtkMultiBlockDataSet. + +To use it: + +* Load the module in Paraview: Tools>Manage Plugins...>Load new>PVGeomechanicsWorkflow. +* Select the Geos output .pvd file loaded in Paraview. +* Select Filters > 3- Geos Geomechanics > Geos Geomechanics Workflow. +* Change the physical constants if needed +* Select computeAdvancedProperties to compute the advanced properties on volume mesh +* Apply PVGeomechanicsWorkflow + +""" + +loggerTitle: str = "GEOS Geomechanics Workflow" + + +@smproxy.filter( + name="PVGeomechanicsWorkflow", + label="Geos Geomechanics Workflow", +) +@smproperty.xml( """ + + + + + + + + + + """ ) +@smproperty.input( name="Input", port_index=0 ) +@smdomain.datatype( dataTypes=[ "vtkMultiBlockDataSet" ], composite_data_supported=True ) +class PVGeomechanicsWorkflow( VTKPythonAlgorithmBase ): + + def __init__( self: Self ) -> None: + """Paraview plugin to compute geomechanics properties on volume and on surface directly from the GEOS simulation output mesh. + + This plugin is the combination of three other: + - First PVGeosBlockExtractAndMerge + - Secondly PVGeomechanicsCalculator + - Thirdly PVSurfaceGeomechanics (if the input mesh contains faults) + """ + super().__init__( + nInputPorts=1, + nOutputPorts=3, + inputType="vtkMultiBlockDataSet", + outputType="vtkMultiBlockDataSet", + ) + + self.volumeMesh: vtkMultiBlockDataSet + self.faultMesh: vtkMultiBlockDataSet + self.wellMesh: vtkMultiBlockDataSet + + self.extractFault: bool = True + self.extractWell: bool = True + + self.computeAdvancedProperties: bool = False + + # Defaults physical constants + ## For basic properties on Volume + self.grainBulkModulus: float = DEFAULT_GRAIN_BULK_MODULUS + self.specificDensity: float = WATER_DENSITY + ## For advanced properties on Volume and basic properties on Surface + self.rockCohesion: float = DEFAULT_ROCK_COHESION + self.frictionAngle: float = DEFAULT_FRICTION_ANGLE_DEG + + self.logger = logging.getLogger( loggerTitle ) + self.logger.setLevel( logging.INFO ) + self.logger.addHandler( VTKHandler() ) + + self.logger.info( f"Apply plugin { self.logger.name }." ) + + @smproperty.doublevector( + name="GrainBulkModulus", + label="Grain bulk modulus (Pa)", + default_values=DEFAULT_GRAIN_BULK_MODULUS, + panel_visibility="default", + ) + @smdomain.xml( """ + + Reference grain bulk modulus to compute Biot coefficient. + The unit is Pa. Default is Quartz bulk modulus (i.e., 38GPa). + + """ ) + def setGrainBulkModulus( self: Self, grainBulkModulus: float ) -> None: + """Set grain bulk modulus. + + Args: + grainBulkModulus (float): Grain bulk modulus (Pa). + """ + self.grainBulkModulus = grainBulkModulus + self.Modified() + + @smproperty.doublevector( + name="SpecificDensity", + label="Specific Density (kg/m3)", + default_values=WATER_DENSITY, + panel_visibility="default", + ) + @smdomain.xml( """ + + Reference density to compute specific gravity. + The unit is kg/m3. Default is fresh water density (i.e., 1000 kg/m3). + + """ ) + def setSpecificDensity( self: Self, specificDensity: float ) -> None: + """Set specific density. + + Args: + specificDensity (float): Reference specific density (kg/m3). + """ + self.specificDensity = specificDensity + self.Modified() + + @smproperty.xml( """ + + + + + """ ) + def groupBasicPropertiesParameters( self: Self ) -> None: + """Organize groups.""" + self.Modified() + + @smproperty.doublevector( + name="RockCohesion", + label="Rock Cohesion (Pa)", + default_values=DEFAULT_ROCK_COHESION, + panel_visibility="default", + ) + @smdomain.xml( """ + + Reference rock cohesion to compute critical pore pressure. + The unit is Pa.Default is fractured case (i.e., 0. Pa). + + """ ) + def setRockCohesion( self: Self, rockCohesion: float ) -> None: + """Set rock cohesion. + + Args: + rockCohesion (float): Rock cohesion (Pa). + """ + self.rockCohesion = rockCohesion + self.Modified() + + @smproperty.doublevector( + name="FrictionAngle", + label="Friction Angle (°)", + default_values=DEFAULT_FRICTION_ANGLE_DEG, + panel_visibility="default", + ) + @smdomain.xml( """ + + Reference friction angle to compute critical pore pressure. + The unit is °. Default is an average friction angle (i.e., 10°). + + """ ) + def setFrictionAngle( self: Self, frictionAngle: float ) -> None: + """Set friction angle. + + Args: + frictionAngle (float): Friction angle (°). + """ + self.frictionAngle = frictionAngle + self.Modified() + + @smproperty.xml( """ + + + + + """ ) + def groupAdvancedPropertiesAndSurfaceParameters( self: Self ) -> None: + """Organize groups.""" + self.Modified() + + @smproperty.intvector( + name="ComputeAdvancedProperties", + label="Compute advanced geomechanics properties", + default_values=0, + panel_visibility="default", + ) + @smdomain.xml( """ + + + Check to compute advanced geomechanics properties including + reservoir stress paths and fracture indexes. + + """ ) + def setComputeAdvancedProperties( self: Self, computeAdvancedProperties: bool ) -> None: + """Set advanced properties calculation option. + + Args: + computeAdvancedProperties (bool): True to compute advanced geomechanics properties, False otherwise. + """ + self.computeAdvancedProperties = computeAdvancedProperties + self.Modified() + + 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 ) + assert inData is not None + + outDataCells = self.GetOutputData( outInfoVec, 0 ) + if outDataCells is None or ( not outDataCells.IsA( "vtkMultiBlockDataSet" ) ): + outDataCells = vtkMultiBlockDataSet() + outInfoVec.GetInformationObject( 0 ).Set( outDataCells.DATA_OBJECT(), outDataCells ) # type: ignore + + outDataFaults = self.GetOutputData( outInfoVec, 1 ) + if outDataFaults is None or ( not outDataFaults.IsA( "vtkMultiBlockDataSet" ) ): + outDataFaults = vtkMultiBlockDataSet() + outInfoVec.GetInformationObject( 1 ).Set( outDataFaults.DATA_OBJECT(), outDataFaults ) # type: ignore + + outDataWells = self.GetOutputData( outInfoVec, 2 ) + if outDataWells is None or ( not outDataWells.IsA( "vtkMultiBlockDataSet" ) ): + outDataWells = vtkMultiBlockDataSet() + outInfoVec.GetInformationObject( 2 ).Set( outDataWells.DATA_OBJECT(), outDataWells ) # type: ignore + + return super().RequestDataObject( request, inInfoVec, outInfoVec ) # type: ignore[no-any-return] + + def RequestData( + self: Self, + request: vtkInformation, + 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. + """ + try: + self.volumeMesh = self.GetOutputData( outInfoVec, 0 ) + self.faultMesh = self.GetOutputData( outInfoVec, 1 ) + self.wellMesh = self.GetOutputData( outInfoVec, 2 ) + + self.applyPVGeosBlockExtractAndMerge() + self.applyPVGeomechanicsCalculator() + + if self.extractFault: + self.applyPVSurfaceGeomechanics() + + self.logger.info( f"The plugin { self.logger.name } succeeded." ) + + except AssertionError as e: + mess: str = "Geomechanics workflow failed due to:" + self.logger.error( mess ) + self.logger.error( str( e ) ) + return 0 + except Exception as e: + mess1: str = "Geomechanics workflow failed due to:" + self.logger.critical( mess1 ) + self.logger.critical( e, exc_info=True ) + return 0 + return 1 + + def applyPVGeosBlockExtractAndMerge( self: Self ) -> None: + """Apply PVGeosBlockExtractAndMerge.""" + extractAndMergeFilter: PVGeosBlockExtractAndMerge = PVGeosBlockExtractAndMerge() + extractAndMergeFilter.SetInputConnection( self.GetInputConnection( 0, 0 ) ) + extractAndMergeFilter.Update() + + self.volumeMesh.ShallowCopy( extractAndMergeFilter.GetOutputDataObject( 0 ) ) + self.volumeMesh.Modified() + + self.extractFault = extractAndMergeFilter.extractFault + if self.extractFault: + self.faultMesh.ShallowCopy( extractAndMergeFilter.GetOutputDataObject( 1 ) ) + self.faultMesh.Modified() + + self.extractWell = extractAndMergeFilter.extractWell + if self.extractWell: + self.wellMesh.ShallowCopy( extractAndMergeFilter.GetOutputDataObject( 2 ) ) + self.wellMesh.Modified() + + return + + def applyPVGeomechanicsCalculator( self: Self ) -> None: + """Apply PVGeomechanicsCalculator.""" + geomechanicsCalculatorPlugin = PVGeomechanicsCalculator() + + geomechanicsCalculatorPlugin.SetInputDataObject( self.volumeMesh ), + geomechanicsCalculatorPlugin.setComputeAdvancedProperties( self.computeAdvancedProperties ) + geomechanicsCalculatorPlugin.setGrainBulkModulus( self.grainBulkModulus ) + geomechanicsCalculatorPlugin.setSpecificDensity( self.specificDensity ) + geomechanicsCalculatorPlugin.setRockCohesion( self.rockCohesion ) + geomechanicsCalculatorPlugin.setFrictionAngle( self.frictionAngle ) + geomechanicsCalculatorPlugin.Update() + + self.volumeMesh.ShallowCopy( geomechanicsCalculatorPlugin.GetOutputDataObject( 0 ) ) + self.volumeMesh.Modified() + + return + + def applyPVSurfaceGeomechanics( self: Self ) -> None: + """Apply PVSurfaceGeomechanics.""" + surfaceGeomechanicsPlugin = PVSurfaceGeomechanics() + + surfaceGeomechanicsPlugin.SetInputDataObject( self.faultMesh ) + surfaceGeomechanicsPlugin.a01SetRockCohesion( self.rockCohesion ) + surfaceGeomechanicsPlugin.a02SetFrictionAngle( self.frictionAngle ) + surfaceGeomechanicsPlugin.Update() + + self.faultMesh.ShallowCopy( surfaceGeomechanicsPlugin.GetOutputDataObject( 0 ) ) + self.faultMesh.Modified() + + return diff --git a/geos-pv/src/geos/pv/plugins/PVGeomechanicsWorkflowVolume.py b/geos-pv/src/geos/pv/plugins/PVGeomechanicsWorkflowVolume.py deleted file mode 100644 index 7c550c37..00000000 --- a/geos-pv/src/geos/pv/plugins/PVGeomechanicsWorkflowVolume.py +++ /dev/null @@ -1,382 +0,0 @@ -# SPDX-License-Identifier: Apache-2.0 -# SPDX-FileCopyrightText: Copyright 2023-2024 TotalEnergies. -# SPDX-FileContributor: Martin Lemay -# ruff: noqa: E402 # disable Module level import not at top of file -import sys -from pathlib import Path - -import numpy as np -from typing_extensions import Self -from vtkmodules.vtkCommonCore import vtkInformation, vtkInformationVector -from vtkmodules.vtkCommonDataModel import ( - vtkMultiBlockDataSet, ) - -# update sys.path to load all GEOS Python Package dependencies -geos_pv_path: Path = Path( __file__ ).parent.parent.parent.parent.parent -sys.path.insert( 0, str( geos_pv_path / "src" ) ) -from geos.pv.utils.config import update_paths - -update_paths() - -from geos.utils.Logger import Logger, getLogger -from geos.utils.PhysicalConstants import ( - DEFAULT_FRICTION_ANGLE_DEG, - DEFAULT_FRICTION_ANGLE_RAD, - DEFAULT_GRAIN_BULK_MODULUS, - DEFAULT_ROCK_COHESION, - WATER_DENSITY, -) -from paraview.util.vtkAlgorithm import ( # type: ignore[import-not-found] - VTKPythonAlgorithmBase, smdomain, smhint, smproperty, smproxy, -) - -from geos.pv.plugins.PVGeosBlockExtractAndMerge import PVGeosBlockExtractAndMerge -from geos.pv.plugins.PVGeomechanicsCalculator import PVGeomechanicsCalculator - -__doc__ = """ -PVGeomechanicsWorkflowVolume is a Paraview plugin that execute multiple filters -to clean GEOS outputs and compute additional geomechanical outputs on volume. - -Input and output types are vtkMultiBlockDataSet. - -This filter results in the volume mesh. If multiple regions were defined in -the volume mesh, they are preserved as distinct blocks. - -To use it: - -* Load the module in Paraview: Tools>Manage Plugins...>Load new>PVGeomechanicsWorkflowVolume. -* Select the Geos output .pvd file loaded in Paraview. -* Search and Apply PVGeomechanicsWorkflowVolume Filter. - -""" - - -@smproxy.filter( - name="PVGeomechanicsWorkflowVolume", - label="Geos Geomechanics Workflow - Volume only", -) -@smhint.xml( """ - - - """ ) -@smproperty.input( name="Input", port_index=0 ) -@smdomain.datatype( dataTypes=[ "vtkMultiBlockDataSet" ], composite_data_supported=True ) -class PVGeomechanicsWorkflowVolume( VTKPythonAlgorithmBase ): - - def __init__( self: Self ) -> None: - """Paraview plugin to clean and add new outputs Geos output mesh. - - To apply in the case of output ".pvd" file contains only Volume element. - """ - super().__init__( - nInputPorts=1, - nOutputPorts=1, - inputType="vtkMultiBlockDataSet", - outputType="vtkMultiBlockDataSet", - ) - - #: ouput volume mesh - self.m_volumeMesh: vtkMultiBlockDataSet - - self.m_computeAdvancedOutputs: bool = False - self.m_grainBulkModulus: float = DEFAULT_GRAIN_BULK_MODULUS - self.m_specificDensity: float = WATER_DENSITY - self.m_rockCohesion: float = DEFAULT_ROCK_COHESION - self.m_frictionAngle: float = DEFAULT_FRICTION_ANGLE_RAD - - # set logger - self.m_logger: Logger = getLogger( "Geomechanics Workflow Filter" ) - - @smproperty.doublevector( - name="GrainBulkModulus", - label="Grain bulk modulus (Pa)", - default_values=DEFAULT_GRAIN_BULK_MODULUS, - panel_visibility="default", - ) - @smdomain.xml( """ - - Reference grain bulk modulus to compute Biot coefficient. - The unit is Pa. Default is Quartz bulk modulus (i.e., 38GPa). - - """ ) - def b01SetGrainBulkModulus( self: Self, value: float ) -> None: - """Set grain bulk modulus. - - Args: - value (float): grain bulk modulus (Pa) - """ - self.m_grainBulkModulus = value - self.Modified() - - def getGrainBulkModulus( self: Self ) -> float: - """Access to the grain bulk modulus value. - - Returns: - float: self.m_grainBulkModulus. - """ - return self.m_grainBulkModulus - - @smproperty.doublevector( - name="SpecificDensity", - label="Specific Density (kg/m3)", - default_values=WATER_DENSITY, - panel_visibility="default", - ) - @smdomain.xml( """ - - Reference density to compute specific gravity. - The unit is kg/m3. Default is fresh water density (i.e., 1000 kg/m3). - - """ ) - def b02SetSpecificDensity( self: Self, value: float ) -> None: - """Set specific density. - - Args: - value (float): Reference specific density (kg/m3) - """ - self.m_specificDensity = value - self.Modified() - - def getSpecificDensity( self: Self ) -> float: - """Access the specific density value. - - Returns: - float: self.m_specificDensity. - """ - return self.m_specificDensity - - @smproperty.xml( """ - - - - - """ ) - def b09GroupBasicOutputParameters( self: Self ) -> None: - """Organize groups.""" - self.Modified() - - @smproperty.intvector( - name="AdvancedOutputsUse", - label="Compute advanced geomechanical outputs", - default_values=0, - panel_visibility="default", - ) - @smdomain.xml( """ - - - Check to compute advanced geomechanical outputs including - reservoir stress paths and fracture indexes. - - """ ) - def c01SetAdvancedOutputs( self: Self, boolean: bool ) -> None: - """Set advanced output calculation option. - - Args: - boolean (bool): if True advanced outputs are computed. - """ - self.m_computeAdvancedOutputs = boolean - self.Modified() - - def getComputeAdvancedOutputs( self: Self ) -> float: - """Access the advanced outputs option. - - Returns: - float: self.m_computeAdvancedOutputs. - """ - return self.m_computeAdvancedOutputs - - @smproperty.xml( """ - - panel_visibility="default"> - - """ ) - def c09GroupAdvancedOutputsUse( self: Self ) -> None: - """Organize groups.""" - self.Modified() - - @smproperty.doublevector( - name="RockCohesion", - label="Rock Cohesion (Pa)", - default_values=DEFAULT_ROCK_COHESION, - panel_visibility="default", - ) - @smdomain.xml( """ - - Reference rock cohesion to compute critical pore pressure. - The unit is Pa.Default is fractured case (i.e., 0. Pa). - - """ ) - def d01SetRockCohesion( self: Self, value: float ) -> None: - """Set rock cohesion. - - Args: - value (float): rock cohesion (Pa) - """ - self.m_rockCohesion = value - self.Modified() - - def getRockCohesion( self: Self ) -> float: - """Get rock cohesion. - - Returns: - float: rock cohesion. - """ - return self.m_rockCohesion - - @smproperty.doublevector( - name="FrictionAngle", - label="Friction Angle (°)", - default_values=DEFAULT_FRICTION_ANGLE_DEG, - panel_visibility="default", - ) - @smdomain.xml( """ - - Reference friction angle to compute critical pore pressure. - The unit is °. Default is 10°. - - """ ) - def d02SetFrictionAngle( self: Self, value: float ) -> None: - """Set frition angle. - - Args: - value (float): friction angle (°) - """ - self.m_frictionAngle = value * np.pi / 180.0 - self.Modified() - - def getFrictionAngle( self: Self ) -> float: - """Get friction angle in radian. - - Returns: - float: friction angle. - """ - return self.m_frictionAngle - - @smproperty.xml( """ - - - - - - - - """ ) - def d09GroupAdvancedOutputParameters( self: Self ) -> None: - """Organize groups.""" - self.Modified() - - def FillInputPortInformation( self: Self, port: int, info: vtkInformation ) -> int: - """Inherited from VTKPythonAlgorithmBase::RequestInformation. - - Args: - port (int): input port - info (vtkInformationVector): info - - Returns: - int: 1 if calculation successfully ended, 0 otherwise. - """ - if port == 0: - info.Set( self.INPUT_REQUIRED_DATA_TYPE(), "vtkMultiBlockDataSet" ) - return 1 - - def RequestInformation( - self: Self, - request: vtkInformation, # noqa: F841 - inInfoVec: list[ vtkInformationVector ], # noqa: F841 - outInfoVec: vtkInformationVector, - ) -> int: - """Inherited from VTKPythonAlgorithmBase::RequestInformation. - - Args: - request (vtkInformation): request - inInfoVec (list[vtkInformationVector]): input objects - outInfoVec (vtkInformationVector): output objects - - Returns: - int: 1 if calculation successfully ended, 0 otherwise. - """ - executive = self.GetExecutive() # noqa: F841 - outInfo: vtkInformation = outInfoVec.GetInformationObject( 0 ) # noqa: F841 - return 1 - - def RequestData( - self: Self, - request: vtkInformation, - 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. - """ - try: - input: vtkMultiBlockDataSet = vtkMultiBlockDataSet.GetData( inInfoVec[ 0 ] ) - self.m_volumeMesh = self.GetOutputData( outInfoVec, 0 ) - - assert input is not None, "Input MultiBlockDataSet is null." - assert self.m_volumeMesh is not None, "Output volume mesh is null." - - # 1. extract volume - self.doExtractAndMerge() - # 2. compute Geomechanical outputs in volume mesh - self.computeAdditionalOutputsVolume() - - except AssertionError as e: - mess: str = "Geomechanics workflow failed due to:" - self.m_logger.error( mess ) - self.m_logger.error( str( e ) ) - return 0 - except Exception as e: - mess1: str = "Geomechanics workflow failed due to:" - self.m_logger.critical( mess1 ) - self.m_logger.critical( e, exc_info=True ) - return 0 - return 1 - - def doExtractAndMerge( self: Self ) -> bool: - """Apply block extraction and merge filter. - - Args: - input (vtkMultiBlockDataSet): input multi block - - Returns: - bool: True if extraction and merge successfully eneded, False otherwise - """ - filter: PVGeosBlockExtractAndMerge = PVGeosBlockExtractAndMerge() - filter.SetInputConnection( self.GetInputConnection( 0, 0 ) ) - filter.SetLogger( self.m_logger ) - filter.Update() - - # recover output objects from PVGeosBlockExtractAndMerge - self.m_volumeMesh.ShallowCopy( filter.GetOutputDataObject( 0 ) ) - self.m_volumeMesh.Modified() - return True - - def computeAdditionalOutputsVolume( self: Self ) -> bool: - """Compute geomechanical outputs on the volume mesh. - - Returns: - bool: True if calculation successfully eneded, False otherwise. - """ - filter = PVGeomechanicsCalculator() - filter.SetInputDataObject( self.m_volumeMesh ), - filter.setComputeAdvancedProperties( self.getComputeAdvancedOutputs() ) - filter.setGrainBulkModulus( self.m_grainBulkModulus ) - filter.setSpecificDensity = ( self.m_specificDensity ) - filter.setRockCohesion = ( self.m_rockCohesion ) - filter.setFrictionAngle = ( self.m_frictionAngle ) - filter.Update() - self.m_volumeMesh.ShallowCopy( filter.GetOutputDataObject( 0 ) ) - self.m_volumeMesh.Modified() - return True diff --git a/geos-pv/src/geos/pv/plugins/PVGeomechanicsWorkflowVolumeSurface.py b/geos-pv/src/geos/pv/plugins/PVGeomechanicsWorkflowVolumeSurface.py deleted file mode 100644 index d383c457..00000000 --- a/geos-pv/src/geos/pv/plugins/PVGeomechanicsWorkflowVolumeSurface.py +++ /dev/null @@ -1,398 +0,0 @@ -# SPDX-License-Identifier: Apache-2.0 -# SPDX-FileCopyrightText: Copyright 2023-2024 TotalEnergies. -# SPDX-FileContributor: Martin Lemay -# ruff: noqa: E402 # disable Module level import not at top of file -import sys -from pathlib import Path - -import numpy as np -from typing_extensions import Self -from vtkmodules.vtkCommonCore import vtkInformation, vtkInformationVector -from vtkmodules.vtkCommonDataModel import ( - vtkMultiBlockDataSet, ) - -# update sys.path to load all GEOS Python Package dependencies -geos_pv_path: Path = Path( __file__ ).parent.parent.parent.parent.parent -sys.path.insert( 0, str( geos_pv_path / "src" ) ) -from geos.pv.utils.config import update_paths - -update_paths() - -from paraview.util.vtkAlgorithm import ( # type: ignore[import-not-found] - VTKPythonAlgorithmBase, smdomain, smhint, smproperty, smproxy, -) - -from geos.utils.Logger import Logger, getLogger -from geos.utils.PhysicalConstants import ( - DEFAULT_FRICTION_ANGLE_DEG, - DEFAULT_FRICTION_ANGLE_RAD, - DEFAULT_GRAIN_BULK_MODULUS, - DEFAULT_ROCK_COHESION, - WATER_DENSITY, -) -from geos.pv.plugins.PVGeosBlockExtractAndMerge import PVGeosBlockExtractAndMerge -from geos.pv.plugins.PVGeomechanicsCalculator import PVGeomechanicsCalculator -from geos.pv.plugins.PVSurfaceGeomechanics import PVSurfaceGeomechanics - -__doc__ = """ -PVGeomechanicsWorkflowVolumeSurface is a Paraview plugin that execute -multiple filters to clean GEOS outputs and compute additional geomechanical -outputs on volume and surface. - -Input and output types are vtkMultiBlockDataSet. - -This filter results in 2 output pipelines: - -* first pipeline contains the volume mesh. If multiple regions were defined in - the volume mesh, they are preserved as distinct blocks. -* second pipeline contains surfaces. If multiple surfaces were used, they are - preserved as distinct blocks. - -To use it: - -* Load the module in Paraview: Tools>Manage Plugins...>Load new>PVGeomechanicsWorkflowVolumeSurface. -* Select the Geos output .pvd file loaded in Paraview. -* Search and Apply PVGeomechanicsWorkflowVolumeSurface Filter. - -""" - - -@smproxy.filter( - name="PVGeomechanicsWorkflowVolumeSurface", - label="Geos Geomechanics Workflow - Volume/Surface", -) -@smhint.xml( """ - - - - """ ) -@smproperty.input( name="Input", port_index=0 ) -@smdomain.datatype( dataTypes=[ "vtkMultiBlockDataSet" ], composite_data_supported=True ) -class PVGeomechanicsWorkflowVolumeSurface( VTKPythonAlgorithmBase ): - - def __init__( self: Self ) -> None: - """Paraview plugin to clean and add new outputs Geos output mesh. - - To apply in the case of output ".pvd" file contains Volume and Fault - elements. - """ - super().__init__( - nInputPorts=1, - nOutputPorts=2, - inputType="vtkMultiBlockDataSet", - outputType="vtkMultiBlockDataSet", - ) - - #: ouput volume mesh - self.m_volumeMesh: vtkMultiBlockDataSet - #: output surface mesh - self.m_surfaceMesh: vtkMultiBlockDataSet - - self.m_computeAdvancedOutputs: bool = False - self.m_grainBulkModulus: float = DEFAULT_GRAIN_BULK_MODULUS - self.m_specificDensity: float = WATER_DENSITY - self.m_rockCohesion: float = DEFAULT_ROCK_COHESION - self.m_frictionAngle: float = DEFAULT_FRICTION_ANGLE_RAD - - # set logger - self.m_logger: Logger = getLogger( "Geomechanics Workflow Filter" ) - - @smproperty.doublevector( - name="GrainBulkModulus", - label="Grain bulk modulus (Pa)", - default_values=DEFAULT_GRAIN_BULK_MODULUS, - panel_visibility="default", - ) - @smdomain.xml( """ - - Reference grain bulk modulus to compute Biot coefficient. - The unit is Pa. Default is Quartz bulk modulus (i.e., 38GPa). - - """ ) - def b01SetGrainBulkModulus( self: Self, value: float ) -> None: - """Set grain bulk modulus. - - Args: - value (float): grain bulk modulus (Pa) - """ - self.m_grainBulkModulus = value - self.Modified() - - def getGrainBulkModulus( self: Self ) -> float: - """Access to the grain bulk modulus value. - - Returns: - float: self.m_grainBulkModulus. - """ - return self.m_grainBulkModulus - - @smproperty.doublevector( - name="SpecificDensity", - label="Specific Density (kg/m3)", - default_values=WATER_DENSITY, - panel_visibility="default", - ) - @smdomain.xml( """ - - Reference density to compute specific gravity. - The unit is kg/m3. Default is fresh water density (i.e., 1000 kg/m3). - - """ ) - def b02SetSpecificDensity( self: Self, value: float ) -> None: - """Set specific density. - - Args: - value (float): Reference specific density (kg/m3) - """ - self.m_specificDensity = value - self.Modified() - - def getSpecificDensity( self: Self ) -> float: - """Access the specific density value. - - Returns: - float: self.m_specificDensity. - """ - return self.m_specificDensity - - @smproperty.xml( """ - - - - - """ ) - def b09GroupBasicOutputParameters( self: Self ) -> None: - """Organize groups.""" - self.Modified() - - @smproperty.doublevector( - name="RockCohesion", - label="Rock Cohesion (Pa)", - default_values=DEFAULT_ROCK_COHESION, - panel_visibility="default", - ) - @smdomain.xml( """ - - Reference rock cohesion to compute critical pore pressure. - The unit is Pa.Default is fractured case (i.e., 0. Pa). - - """ ) - def d01SetRockCohesion( self: Self, value: float ) -> None: - """Set rock cohesion. - - Args: - value (float): rock cohesion (Pa) - """ - self.m_rockCohesion = value - self.Modified() - - def getRockCohesion( self: Self ) -> float: - """Get rock cohesion. - - Returns: - float: rock cohesion. - """ - return self.m_rockCohesion - - @smproperty.doublevector( - name="FrictionAngle", - label="Friction Angle (°)", - default_values=DEFAULT_FRICTION_ANGLE_DEG, - panel_visibility="default", - ) - @smdomain.xml( """ - - Reference friction angle to compute critical pore pressure. - The unit is °. Default is 10°. - - """ ) - def d02SetFrictionAngle( self: Self, value: float ) -> None: - """Set frition angle. - - Args: - value (float): friction angle (°) - """ - self.m_frictionAngle = value * np.pi / 180.0 - self.Modified() - - def getFrictionAngle( self: Self ) -> float: - """Get friction angle in radian. - - Returns: - float: friction angle. - """ - return self.m_frictionAngle - - @smproperty.xml( """ - - - - - """ ) - def d09GroupAdvancedOutputParameters( self: Self ) -> None: - """Organize groups.""" - self.Modified() - - @smproperty.intvector( - name="AdvancedOutputsUse", - label="Compute advanced geomechanical volume outputs", - default_values=0, - panel_visibility="default", - ) - @smdomain.xml( """ - - - Check to compute advanced geomechanical outputs including - reservoir stress paths and fracture indexes. - - """ ) - def e01SetAdvancedOutputs( self: Self, boolean: bool ) -> None: - """Set advanced output calculation option. - - Args: - boolean (bool): if True advanced outputs are computed. - """ - self.m_computeAdvancedOutputs = boolean - self.Modified() - - def getComputeAdvancedOutputs( self: Self ) -> float: - """Access the advanced outputs option. - - Returns: - float: self.m_computeAdvancedOutputs. - """ - return self.m_computeAdvancedOutputs - - def FillInputPortInformation( self: Self, port: int, info: vtkInformation ) -> int: - """Inherited from VTKPythonAlgorithmBase::RequestInformation. - - Args: - port (int): input port - info (vtkInformationVector): info - - Returns: - int: 1 if calculation successfully ended, 0 otherwise. - """ - if port == 0: - info.Set( self.INPUT_REQUIRED_DATA_TYPE(), "vtkMultiBlockDataSet" ) - return 1 - - def RequestInformation( - self: Self, - request: vtkInformation, # noqa: F841 - inInfoVec: list[ vtkInformationVector ], # noqa: F841 - outInfoVec: vtkInformationVector, - ) -> int: - """Inherited from VTKPythonAlgorithmBase::RequestInformation. - - Args: - request (vtkInformation): request - inInfoVec (list[vtkInformationVector]): input objects - outInfoVec (vtkInformationVector): output objects - - Returns: - int: 1 if calculation successfully ended, 0 otherwise. - """ - executive = self.GetExecutive() # noqa: F841 - outInfo: vtkInformation = outInfoVec.GetInformationObject( 0 ) # noqa: F841 - return 1 - - def RequestData( - self: Self, - request: vtkInformation, - 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. - """ - try: - input: vtkMultiBlockDataSet = vtkMultiBlockDataSet.GetData( inInfoVec[ 0 ] ) - self.m_volumeMesh = self.GetOutputData( outInfoVec, 0 ) - self.m_surfaceMesh = self.GetOutputData( outInfoVec, 1 ) - - assert input is not None, "Input MultiBlockDataSet is null." - assert self.m_volumeMesh is not None, "Output volume mesh is null." - assert self.m_surfaceMesh is not None, "Output surface mesh is null." - - # 1. extract volume/surface - self.doExtractAndMerge() - # 2. compute Geomechanical outputs in volume mesh - self.computeAdditionalOutputsVolume() - # 3. compute geomechanical outputs on surface mesh - self.computeSurfaceGeomecanics() - - except AssertionError as e: - mess: str = "Geomechanics workflow failed due to:" - self.m_logger.error( mess ) - self.m_logger.error( str( e ) ) - return 0 - except Exception as e: - mess1: str = "Geomechanics workflow failed due to:" - self.m_logger.critical( mess1 ) - self.m_logger.critical( e, exc_info=True ) - return 0 - return 1 - - def doExtractAndMerge( self: Self ) -> bool: - """Apply block extraction and merge filter. - - Args: - input (vtkMultiBlockDataSet): input multi block - - Returns: - bool: True if extraction and merge successfully eneded, False otherwise - """ - filter: PVGeosBlockExtractAndMerge = PVGeosBlockExtractAndMerge() - filter.SetInputConnection( self.GetInputConnection( 0, 0 ) ) - filter.SetLogger( self.m_logger ) - filter.Update() - - # recover output objects from PVGeosBlockExtractAndMerge - self.m_volumeMesh.ShallowCopy( filter.GetOutputDataObject( 0 ) ) - self.m_surfaceMesh.ShallowCopy( filter.GetOutputDataObject( 1 ) ) - self.m_volumeMesh.Modified() - self.m_surfaceMesh.Modified() - return True - - def computeAdditionalOutputsVolume( self: Self ) -> bool: - """Compute geomechanical outputs on the volume mesh. - - Returns: - bool: True if calculation successfully eneded, False otherwise. - """ - filter = PVGeomechanicsCalculator() - filter.SetInputDataObject( self.m_volumeMesh ), - filter.setComputeAdvancedProperties( self.getComputeAdvancedOutputs() ) - filter.setGrainBulkModulus( self.m_grainBulkModulus ) - filter.setSpecificDensity = ( self.m_specificDensity ) - filter.setRockCohesion = ( self.m_rockCohesion ) - filter.setFrictionAngle = ( self.m_frictionAngle ) - filter.Update() - self.m_volumeMesh.ShallowCopy( filter.GetOutputDataObject( 0 ) ) - self.m_volumeMesh.Modified() - return True - - def computeSurfaceGeomecanics( self: Self ) -> bool: - """Compute surface geomechanics new attributes. - - Returns: - bool: True if calculation successfully eneded, False otherwise. - """ - filter = PVSurfaceGeomechanics() - filter.SetInputDataObject( self.m_surfaceMesh ) - filter.a01SetRockCohesion( self.getRockCohesion() ) - filter.a02SetFrictionAngle( self.getFrictionAngle() ) - filter.SetLogger( self.m_logger ) - filter.Update() - self.m_surfaceMesh.ShallowCopy( filter.GetOutputDataObject( 0 ) ) - self.m_surfaceMesh.Modified() - return True diff --git a/geos-pv/src/geos/pv/plugins/PVGeomechanicsWorkflowVolumeSurfaceWell.py b/geos-pv/src/geos/pv/plugins/PVGeomechanicsWorkflowVolumeSurfaceWell.py deleted file mode 100644 index 3c8ac8fb..00000000 --- a/geos-pv/src/geos/pv/plugins/PVGeomechanicsWorkflowVolumeSurfaceWell.py +++ /dev/null @@ -1,407 +0,0 @@ -# SPDX-License-Identifier: Apache-2.0 -# SPDX-FileCopyrightText: Copyright 2023-2024 TotalEnergies. -# SPDX-FileContributor: Martin Lemay -# ruff: noqa: E402 # disable Module level import not at top of file -import sys -from pathlib import Path - -import numpy as np -from typing_extensions import Self -from vtkmodules.vtkCommonCore import vtkInformation, vtkInformationVector -from vtkmodules.vtkCommonDataModel import ( - vtkMultiBlockDataSet, ) - -# update sys.path to load all GEOS Python Package dependencies -geos_pv_path: Path = Path( __file__ ).parent.parent.parent.parent.parent -sys.path.insert( 0, str( geos_pv_path / "src" ) ) -from geos.pv.utils.config import update_paths - -update_paths() - -from paraview.util.vtkAlgorithm import ( # type: ignore[import-not-found] - VTKPythonAlgorithmBase, smdomain, smhint, smproperty, smproxy, -) - -from geos.utils.Logger import Logger, getLogger -from geos.utils.PhysicalConstants import ( - DEFAULT_FRICTION_ANGLE_DEG, - DEFAULT_FRICTION_ANGLE_RAD, - DEFAULT_GRAIN_BULK_MODULUS, - DEFAULT_ROCK_COHESION, - WATER_DENSITY, -) -from geos.pv.plugins.PVGeosBlockExtractAndMerge import PVGeosBlockExtractAndMerge -from geos.pv.plugins.PVGeomechanicsCalculator import PVGeomechanicsCalculator -from geos.pv.plugins.PVSurfaceGeomechanics import PVSurfaceGeomechanics - -__doc__ = """ -PVGeomechanicsWorkflowVolumeSurfaceWell is a Paraview plugin that execute -multiple filters to clean GEOS outputs and compute additional geomechanical -outputs on volume, surface and wells. - -Input and output types are vtkMultiBlockDataSet. - -This filter results in 3 output pipelines: - -* first pipeline contains the volume mesh. If multiple regions were defined in - the volume mesh, they are preserved as distinct blocks. -* second pipeline contains surfaces. If multiple surfaces were used, they are - preserved as distinct blocks. -* third pipeline contains wells. If multiple wells were used, they are preserved - as distinct blocks. - -To use it: - -* Load the module in Paraview: Tools>Manage Plugins...>Load new>PVGeomechanicsWorkflowVolumeSurfaceWell. -* Select the Geos output .pvd file loaded in Paraview. -* Search and Apply PVGeomechanicsWorkflowVolumeSurfaceWell Filter. - -""" - - -@smproxy.filter( - name="PVGeomechanicsWorkflowVolumeSurfaceWell", - label="Geos Geomechanics Workflow - Volume/Surface/Well", -) -@smhint.xml( """ - - - - - """ ) -@smproperty.input( name="Input", port_index=0 ) -@smdomain.datatype( dataTypes=[ "vtkMultiBlockDataSet" ], composite_data_supported=True ) -class PVGeomechanicsWorkflowVolumeSurfaceWell( VTKPythonAlgorithmBase ): - - def __init__( self: Self ) -> None: - """Paraview plugin to clean and add new outputs Geos output mesh. - - To apply in the case of output ".pvd" file contains Volume, Fault and - Well elements. - """ - super().__init__( - nInputPorts=1, - nOutputPorts=3, - inputType="vtkMultiBlockDataSet", - outputType="vtkMultiBlockDataSet", - ) - - #: ouput volume mesh - self.m_volumeMesh: vtkMultiBlockDataSet - #: output surface mesh - self.m_surfaceMesh: vtkMultiBlockDataSet - #: output wells - self.m_wells: vtkMultiBlockDataSet - - self.m_computeAdvancedOutputs: bool = False - self.m_grainBulkModulus: float = DEFAULT_GRAIN_BULK_MODULUS - self.m_specificDensity: float = WATER_DENSITY - self.m_rockCohesion: float = DEFAULT_ROCK_COHESION - self.m_frictionAngle: float = DEFAULT_FRICTION_ANGLE_RAD - - # set logger - self.m_logger: Logger = getLogger( "Geomechanics Workflow Filter" ) - - @smproperty.doublevector( - name="GrainBulkModulus", - label="Grain bulk modulus (Pa)", - default_values=DEFAULT_GRAIN_BULK_MODULUS, - panel_visibility="default", - ) - @smdomain.xml( """ - - Reference grain bulk modulus to compute Biot coefficient. - The unit is Pa. Default is Quartz bulk modulus (i.e., 38GPa). - - """ ) - def b01SetGrainBulkModulus( self: Self, value: float ) -> None: - """Set grain bulk modulus. - - Args: - value (float): grain bulk modulus (Pa) - """ - self.m_grainBulkModulus = value - self.Modified() - - def getGrainBulkModulus( self: Self ) -> float: - """Access to the grain bulk modulus value. - - Returns: - float: self.m_grainBulkModulus. - """ - return self.m_grainBulkModulus - - @smproperty.doublevector( - name="SpecificDensity", - label="Specific Density (kg/m3)", - default_values=WATER_DENSITY, - panel_visibility="default", - ) - @smdomain.xml( """ - - Reference density to compute specific gravity. - The unit is kg/m3. Default is fresh water density (i.e., 1000 kg/m3). - - """ ) - def b02SetSpecificDensity( self: Self, value: float ) -> None: - """Set specific density. - - Args: - value (float): Reference specific density (kg/m3) - """ - self.m_specificDensity = value - self.Modified() - - def getSpecificDensity( self: Self ) -> float: - """Access the specific density value. - - Returns: - float: self.m_specificDensity. - """ - return self.m_specificDensity - - @smproperty.xml( """ - - - - - """ ) - def b09GroupBasicOutputParameters( self: Self ) -> None: - """Organize groups.""" - self.Modified() - - @smproperty.doublevector( - name="RockCohesion", - label="Rock Cohesion (Pa)", - default_values=DEFAULT_ROCK_COHESION, - panel_visibility="default", - ) - @smdomain.xml( """ - - Reference rock cohesion to compute critical pore pressure. - The unit is Pa.Default is fractured case (i.e., 0. Pa). - - """ ) - def d01SetRockCohesion( self: Self, value: float ) -> None: - """Set rock cohesion. - - Args: - value (float): rock cohesion (Pa) - """ - self.m_rockCohesion = value - self.Modified() - - def getRockCohesion( self: Self ) -> float: - """Get rock cohesion. - - Returns: - float: rock cohesion. - """ - return self.m_rockCohesion - - @smproperty.doublevector( - name="FrictionAngle", - label="Friction Angle (°)", - default_values=DEFAULT_FRICTION_ANGLE_DEG, - panel_visibility="default", - ) - @smdomain.xml( """ - - Reference friction angle to compute critical pore pressure. - The unit is °. Default is 10°. - - """ ) - def d02SetFrictionAngle( self: Self, value: float ) -> None: - """Set frition angle. - - Args: - value (float): friction angle (°) - """ - self.m_frictionAngle = value * np.pi / 180.0 - self.Modified() - - def getFrictionAngle( self: Self ) -> float: - """Get friction angle in radian. - - Returns: - float: friction angle. - """ - return self.m_frictionAngle - - @smproperty.xml( """ - - - - - """ ) - def d09GroupAdvancedOutputParameters( self: Self ) -> None: - """Organize groups.""" - self.Modified() - - @smproperty.intvector( - name="AdvancedOutputsUse", - label="Compute advanced geomechanical volume outputs", - default_values=0, - panel_visibility="default", - ) - @smdomain.xml( """ - - - Check to compute advanced geomechanical outputs including - reservoir stress paths and fracture indexes. - - """ ) - def e01SetAdvancedOutputs( self: Self, boolean: bool ) -> None: - """Set advanced output calculation option. - - Args: - boolean (bool): if True advanced outputs are computed. - """ - self.m_computeAdvancedOutputs = boolean - self.Modified() - - def getComputeAdvancedOutputs( self: Self ) -> float: - """Access the advanced outputs option. - - Returns: - float: self.m_computeAdvancedOutputs. - """ - return self.m_computeAdvancedOutputs - - def FillInputPortInformation( self: Self, port: int, info: vtkInformation ) -> int: - """Inherited from VTKPythonAlgorithmBase::RequestInformation. - - Args: - port (int): input port - info (vtkInformationVector): info - - Returns: - int: 1 if calculation successfully ended, 0 otherwise. - """ - if port == 0: - info.Set( self.INPUT_REQUIRED_DATA_TYPE(), "vtkMultiBlockDataSet" ) - return 1 - - def RequestInformation( - self: Self, - request: vtkInformation, # noqa: F841 - inInfoVec: list[ vtkInformationVector ], # noqa: F841 - outInfoVec: vtkInformationVector, - ) -> int: - """Inherited from VTKPythonAlgorithmBase::RequestInformation. - - Args: - request (vtkInformation): request - inInfoVec (list[vtkInformationVector]): input objects - outInfoVec (vtkInformationVector): output objects - - Returns: - int: 1 if calculation successfully ended, 0 otherwise. - """ - executive = self.GetExecutive() # noqa: F841 - outInfo: vtkInformation = outInfoVec.GetInformationObject( 0 ) # noqa: F841 - return 1 - - def RequestData( - self: Self, - request: vtkInformation, - 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. - """ - try: - input: vtkMultiBlockDataSet = vtkMultiBlockDataSet.GetData( inInfoVec[ 0 ] ) - self.m_volumeMesh = self.GetOutputData( outInfoVec, 0 ) - self.m_surfaceMesh = self.GetOutputData( outInfoVec, 1 ) - self.m_wells = self.GetOutputData( outInfoVec, 2 ) - - assert input is not None, "Input MultiBlockDataSet is null." - assert self.m_volumeMesh is not None, "Output volume mesh is null." - assert self.m_surfaceMesh is not None, "Output surface mesh is null." - assert self.m_wells is not None, "Output well mesh is null." - - # 1. extract volume/surface/wells - self.doExtractAndMerge() - # 2. compute Geomechanical outputs in volume mesh - self.computeAdditionalOutputsVolume() - # 3. compute geomechanical outputs on surface mesh - self.computeSurfaceGeomecanics() - - except AssertionError as e: - mess: str = "Geomechanics workflow failed due to:" - self.m_logger.error( mess ) - self.m_logger.error( str( e ) ) - return 0 - except Exception as e: - mess1: str = "Geomechanics workflow failed due to:" - self.m_logger.critical( mess1 ) - self.m_logger.critical( e, exc_info=True ) - return 0 - return 1 - - def doExtractAndMerge( self: Self ) -> bool: - """Apply block extraction and merge filter. - - Args: - input (vtkMultiBlockDataSet): input multi block - - Returns: - bool: True if extraction and merge successfully eneded, False otherwise - """ - filter: PVGeosBlockExtractAndMerge = PVGeosBlockExtractAndMerge() - filter.SetInputConnection( self.GetInputConnection( 0, 0 ) ) - filter.SetLogger( self.m_logger ) - filter.Update() - - # recover output objects from PVGeosBlockExtractAndMerge - self.m_volumeMesh.ShallowCopy( filter.GetOutputDataObject( 0 ) ) - self.m_surfaceMesh.ShallowCopy( filter.GetOutputDataObject( 1 ) ) - self.m_wells.ShallowCopy( filter.GetOutputDataObject( 2 ) ) - self.m_volumeMesh.Modified() - self.m_surfaceMesh.Modified() - self.m_wells.Modified() - return True - - def computeAdditionalOutputsVolume( self: Self ) -> bool: - """Compute geomechanical outputs on the volume mesh. - - Returns: - bool: True if calculation successfully eneded, False otherwise. - """ - filter = PVGeomechanicsCalculator() - filter.SetInputDataObject( self.m_volumeMesh ), - filter.setComputeAdvancedProperties( self.getComputeAdvancedOutputs() ) - filter.setGrainBulkModulus( self.m_grainBulkModulus ) - filter.setSpecificDensity = ( self.m_specificDensity ) - filter.setRockCohesion = ( self.m_rockCohesion ) - filter.setFrictionAngle = ( self.m_frictionAngle ) - filter.Update() - self.m_volumeMesh.ShallowCopy( filter.GetOutputDataObject( 0 ) ) - self.m_volumeMesh.Modified() - return True - - def computeSurfaceGeomecanics( self: Self ) -> bool: - """Compute surface geomechanics new attributes. - - Returns: - bool: True if calculation successfully eneded, False otherwise. - """ - filter = PVSurfaceGeomechanics() - filter.SetInputDataObject( self.m_surfaceMesh ) - filter.a01SetRockCohesion( self.getRockCohesion() ) - filter.a02SetFrictionAngle( self.getFrictionAngle() ) - filter.SetLogger( self.m_logger ) - filter.Update() - self.m_surfaceMesh.ShallowCopy( filter.GetOutputDataObject( 0 ) ) - self.m_surfaceMesh.Modified() - return True diff --git a/geos-pv/src/geos/pv/plugins/PVGeomechanicsWorkflowVolumeWell.py b/geos-pv/src/geos/pv/plugins/PVGeomechanicsWorkflowVolumeWell.py deleted file mode 100644 index 44aeb977..00000000 --- a/geos-pv/src/geos/pv/plugins/PVGeomechanicsWorkflowVolumeWell.py +++ /dev/null @@ -1,396 +0,0 @@ -# SPDX-License-Identifier: Apache-2.0 -# SPDX-FileCopyrightText: Copyright 2023-2024 TotalEnergies. -# SPDX-FileContributor: Martin Lemay -# ruff: noqa: E402 # disable Module level import not at top of file -import sys -from pathlib import Path - -import numpy as np -from typing_extensions import Self -from vtkmodules.vtkCommonCore import vtkInformation, vtkInformationVector -from vtkmodules.vtkCommonDataModel import ( - vtkMultiBlockDataSet, ) - -# update sys.path to load all GEOS Python Package dependencies -geos_pv_path: Path = Path( __file__ ).parent.parent.parent.parent.parent -sys.path.insert( 0, str( geos_pv_path / "src" ) ) -from geos.pv.utils.config import update_paths - -update_paths() - -from paraview.util.vtkAlgorithm import ( # type: ignore[import-not-found] - VTKPythonAlgorithmBase, smdomain, smhint, smproperty, smproxy, -) - -from geos.utils.Logger import Logger, getLogger -from geos.utils.PhysicalConstants import ( - DEFAULT_FRICTION_ANGLE_DEG, - DEFAULT_FRICTION_ANGLE_RAD, - DEFAULT_GRAIN_BULK_MODULUS, - DEFAULT_ROCK_COHESION, - WATER_DENSITY, -) - -from geos.pv.plugins.PVGeosBlockExtractAndMerge import PVGeosBlockExtractAndMerge -from geos.pv.plugins.PVGeomechanicsCalculator import PVGeomechanicsCalculator - -__doc__ = """ -PVGeomechanicsWorkflowVolumeWell is a Paraview plugin that execute -multiple filters to clean GEOS outputs and compute additional geomechanical -outputs on volume and wells. - -Input and output types are vtkMultiBlockDataSet. - -This filter results in 2 output pipelines: - -* first pipeline contains the volume mesh. If multiple regions were defined in - the volume mesh, they are preserved as distinct blocks. -* second pipeline contains wells. If multiple wells were used, they are preserved - as distinct blocks. - -To use it: - -* Load the module in Paraview: Tools>Manage Plugins...>Load new>PVGeomechanicsWorkflowVolumeWell. -* Select the Geos output .pvd file loaded in Paraview. -* Search and Apply PVGeomechanicsWorkflowVolumeWell Filter. - -""" - - -@smproxy.filter( - name="PVGeomechanicsWorkflowVolumeWell", - label="Geos Geomechanics Workflow - Volume/Well", -) -@smhint.xml( """ - - - - """ ) -@smproperty.input( name="Input", port_index=0 ) -@smdomain.datatype( dataTypes=[ "vtkMultiBlockDataSet" ], composite_data_supported=True ) -class PVGeomechanicsWorkflowVolumeWell( VTKPythonAlgorithmBase ): - - def __init__( self: Self ) -> None: - """Paraview plugin to clean and add new outputs Geos output mesh. - - To apply in the case of output ".pvd" file contains Volume and - Well elements. - """ - super().__init__( - nInputPorts=1, - nOutputPorts=2, - inputType="vtkMultiBlockDataSet", - outputType="vtkMultiBlockDataSet", - ) - - #: ouput volume mesh - self.m_volumeMesh: vtkMultiBlockDataSet - #: output wells - self.m_wells: vtkMultiBlockDataSet - - self.m_computeAdvancedOutputs: bool = False - self.m_grainBulkModulus: float = DEFAULT_GRAIN_BULK_MODULUS - self.m_specificDensity: float = WATER_DENSITY - self.m_rockCohesion: float = DEFAULT_ROCK_COHESION - self.m_frictionAngle: float = DEFAULT_FRICTION_ANGLE_RAD - - # set logger - self.m_logger: Logger = getLogger( "Geomechanics Workflow Filter" ) - - @smproperty.doublevector( - name="GrainBulkModulus", - label="Grain bulk modulus (Pa)", - default_values=DEFAULT_GRAIN_BULK_MODULUS, - panel_visibility="default", - ) - @smdomain.xml( """ - - Reference grain bulk modulus to compute Biot coefficient. - The unit is Pa. Default is Quartz bulk modulus (i.e., 38GPa). - - """ ) - def b01SetGrainBulkModulus( self: Self, value: float ) -> None: - """Set grain bulk modulus. - - Args: - value (float): grain bulk modulus (Pa) - """ - self.m_grainBulkModulus = value - self.Modified() - - def getGrainBulkModulus( self: Self ) -> float: - """Access to the grain bulk modulus value. - - Returns: - float: self.m_grainBulkModulus. - """ - return self.m_grainBulkModulus - - @smproperty.doublevector( - name="SpecificDensity", - label="Specific Density (kg/m3)", - default_values=WATER_DENSITY, - panel_visibility="default", - ) - @smdomain.xml( """ - - Reference density to compute specific gravity. - The unit is kg/m3. Default is fresh water density (i.e., 1000 kg/m3). - - """ ) - def b02SetSpecificDensity( self: Self, value: float ) -> None: - """Set specific density. - - Args: - value (float): Reference specific density (kg/m3) - """ - self.m_specificDensity = value - self.Modified() - - def getSpecificDensity( self: Self ) -> float: - """Access the specific density value. - - Returns: - float: self.m_specificDensity. - """ - return self.m_specificDensity - - @smproperty.xml( """ - - - - - """ ) - def b09GroupBasicOutputParameters( self: Self ) -> None: - """Organize groups.""" - self.Modified() - - @smproperty.intvector( - name="AdvancedOutputsUse", - label="Compute advanced geomechanical outputs", - default_values=0, - panel_visibility="default", - ) - @smdomain.xml( """ - - - Check to compute advanced geomechanical outputs including - reservoir stress paths and fracture indexes. - - """ ) - def c01SetAdvancedOutputs( self: Self, boolean: bool ) -> None: - """Set advanced output calculation option. - - Args: - boolean (bool): if True advanced outputs are computed. - """ - self.m_computeAdvancedOutputs = boolean - self.Modified() - - def getComputeAdvancedOutputs( self: Self ) -> float: - """Access the advanced outputs option. - - Returns: - float: self.m_computeAdvancedOutputs. - """ - return self.m_computeAdvancedOutputs - - @smproperty.xml( """ - - panel_visibility="default"> - - """ ) - def c09GroupAdvancedOutputsUse( self: Self ) -> None: - """Organize groups.""" - self.Modified() - - @smproperty.doublevector( - name="RockCohesion", - label="Rock Cohesion (Pa)", - default_values=DEFAULT_ROCK_COHESION, - panel_visibility="default", - ) - @smdomain.xml( """ - - Reference rock cohesion to compute critical pore pressure. - The unit is Pa.Default is fractured case (i.e., 0. Pa). - - """ ) - def d01SetRockCohesion( self: Self, value: float ) -> None: - """Set rock cohesion. - - Args: - value (float): rock cohesion (Pa) - """ - self.m_rockCohesion = value - self.Modified() - - def getRockCohesion( self: Self ) -> float: - """Get rock cohesion. - - Returns: - float: rock cohesion. - """ - return self.m_rockCohesion - - @smproperty.doublevector( - name="FrictionAngle", - label="Friction Angle (°)", - default_values=DEFAULT_FRICTION_ANGLE_DEG, - panel_visibility="default", - ) - @smdomain.xml( """ - - Reference friction angle to compute critical pore pressure. - The unit is °. Default is 10°. - - """ ) - def d02SetFrictionAngle( self: Self, value: float ) -> None: - """Set frition angle. - - Args: - value (float): friction angle (°) - """ - self.m_frictionAngle = value * np.pi / 180.0 - self.Modified() - - def getFrictionAngle( self: Self ) -> float: - """Get friction angle in radian. - - Returns: - float: friction angle. - """ - return self.m_frictionAngle - - @smproperty.xml( """ - - - - - - - - """ ) - def d09GroupAdvancedOutputParameters( self: Self ) -> None: - """Organize groups.""" - self.Modified() - - def FillInputPortInformation( self: Self, port: int, info: vtkInformation ) -> int: - """Inherited from VTKPythonAlgorithmBase::RequestInformation. - - Args: - port (int): input port - info (vtkInformationVector): info - - Returns: - int: 1 if calculation successfully ended, 0 otherwise. - """ - if port == 0: - info.Set( self.INPUT_REQUIRED_DATA_TYPE(), "vtkMultiBlockDataSet" ) - return 1 - - def RequestInformation( - self: Self, - request: vtkInformation, # noqa: F841 - inInfoVec: list[ vtkInformationVector ], # noqa: F841 - outInfoVec: vtkInformationVector, - ) -> int: - """Inherited from VTKPythonAlgorithmBase::RequestInformation. - - Args: - request (vtkInformation): request - inInfoVec (list[vtkInformationVector]): input objects - outInfoVec (vtkInformationVector): output objects - - Returns: - int: 1 if calculation successfully ended, 0 otherwise. - """ - executive = self.GetExecutive() # noqa: F841 - outInfo: vtkInformation = outInfoVec.GetInformationObject( 0 ) # noqa: F841 - return 1 - - def RequestData( - self: Self, - request: vtkInformation, - 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. - """ - try: - input: vtkMultiBlockDataSet = vtkMultiBlockDataSet.GetData( inInfoVec[ 0 ] ) - self.m_volumeMesh = self.GetOutputData( outInfoVec, 0 ) - self.m_wells = self.GetOutputData( outInfoVec, 1 ) - - assert input is not None, "Input MultiBlockDataSet is null." - assert self.m_volumeMesh is not None, "Output volume mesh is null." - assert self.m_wells is not None, "Output well mesh is null." - - # 1. extract volume/wells - self.doExtractAndMerge() - # 2. compute Geomechanical outputs in volume mesh - self.computeAdditionalOutputsVolume() - - except AssertionError as e: - mess: str = "Geomechanics workflow failed due to:" - self.m_logger.error( mess ) - self.m_logger.error( str( e ) ) - return 0 - except Exception as e: - mess1: str = "Geomechanics workflow failed due to:" - self.m_logger.critical( mess1 ) - self.m_logger.critical( e, exc_info=True ) - return 0 - return 1 - - def doExtractAndMerge( self: Self ) -> bool: - """Apply block extraction and merge filter. - - Args: - input (vtkMultiBlockDataSet): input multi block - - Returns: - bool: True if extraction and merge successfully eneded, False otherwise - """ - filter: PVGeosBlockExtractAndMerge = PVGeosBlockExtractAndMerge() - filter.SetInputConnection( self.GetInputConnection( 0, 0 ) ) - filter.SetLogger( self.m_logger ) - filter.Update() - - # recover output objects from PVGeosBlockExtractAndMerge - self.m_volumeMesh.ShallowCopy( filter.GetOutputDataObject( 0 ) ) - self.m_wells.ShallowCopy( filter.GetOutputDataObject( 1 ) ) - self.m_volumeMesh.Modified() - self.m_wells.Modified() - return True - - def computeAdditionalOutputsVolume( self: Self ) -> bool: - """Compute geomechanical outputs on the volume mesh. - - Returns: - bool: True if calculation successfully eneded, False otherwise. - """ - filter = PVGeomechanicsCalculator() - filter.SetInputDataObject( self.m_volumeMesh ), - filter.setComputeAdvancedProperties( self.getComputeAdvancedOutputs() ) - filter.setGrainBulkModulus( self.m_grainBulkModulus ) - filter.setSpecificDensity = ( self.m_specificDensity ) - filter.setRockCohesion = ( self.m_rockCohesion ) - filter.setFrictionAngle = ( self.m_frictionAngle ) - filter.Update() - self.m_volumeMesh.ShallowCopy( filter.GetOutputDataObject( 0 ) ) - self.m_volumeMesh.Modified() - return True diff --git a/geos-pv/src/geos/pv/plugins/PVGeosBlockExtractAndMerge.py b/geos-pv/src/geos/pv/plugins/PVGeosBlockExtractAndMerge.py index beea5400..c779c5eb 100644 --- a/geos-pv/src/geos/pv/plugins/PVGeosBlockExtractAndMerge.py +++ b/geos-pv/src/geos/pv/plugins/PVGeosBlockExtractAndMerge.py @@ -117,6 +117,9 @@ def __init__( self: Self ) -> None: self.logger = logging.getLogger( loggerTitle ) self.logger.setLevel( logging.INFO ) self.logger.addHandler( VTKHandler() ) + self.logger.propagate = False + + self.logger.info( f"Apply plugin { self.logger.name }." ) def RequestDataObject( self: Self, @@ -170,8 +173,6 @@ def RequestInformation( Returns: int: 1 if calculation successfully ended, 0 otherwise. """ - self.logger.info( f"Apply plugin { self.logger.name }." ) - executive = self.GetExecutive() inInfo = inInfoVec[ 0 ] self.timeSteps = inInfo.GetInformationObject( 0 ).Get( executive.TIME_STEPS() ) diff --git a/geos-pv/src/geos/pv/utils/workflowFunctions.py b/geos-pv/src/geos/pv/utils/workflowFunctions.py index 39ae35ee..28602fce 100644 --- a/geos-pv/src/geos/pv/utils/workflowFunctions.py +++ b/geos-pv/src/geos/pv/utils/workflowFunctions.py @@ -3,13 +3,12 @@ # SPDX-FileContributor: Romain Baville # ruff: noqa: E402 # disable Module level import not at top of file - from geos.processing.post_processing.GeosBlockExtractor import GeosBlockExtractor from geos.processing.post_processing.GeosBlockMerge import GeosBlockMerge from vtkmodules.vtkCommonDataModel import vtkMultiBlockDataSet -from paraview.detail.loghandler import ( VTKHandler ) # type: ignore[import-not-found] +from paraview.detail.loghandler import ( VTKHandler ) # type: ignore[import-not-found] def doExtractAndMerge( @@ -23,47 +22,59 @@ def doExtractAndMerge( """Apply block extraction and merge. Args: - mesh (vtkMultiBlockDataSet): Mesh to process. - outputCells (vtkMultiBlockDataSet): Output volume mesh. - outputFaults (vtkMultiBlockDataSet): Output surface mesh. - outputWells (vtkMultiBlockDataSet): Output well mesh. + mesh (vtkMultiBlockDataSet): Mesh to process + outputCells (vtkMultiBlockDataSet): Output volume mesh + outputFaults (vtkMultiBlockDataSet): Output surface mesh + outputWells (vtkMultiBlockDataSet): Output well mesh + extractFault (bool): True if SurfaceElementRegion needs to be extracted, False otherwise. + extractWell (bool): True if WellElementRegion needs to be extracted, False otherwise. """ # Extract blocks - blockExtractor: GeosBlockExtractor = GeosBlockExtractor( mesh, extractFault=extractFault, extractWell=extractWell, speHandler=True ) + blockExtractor: GeosBlockExtractor = GeosBlockExtractor( mesh, + extractFault=extractFault, + extractWell=extractWell, + speHandler=True ) if not blockExtractor.logger.hasHandlers(): blockExtractor.setLoggerHandler( VTKHandler() ) blockExtractor.applyFilter() # recover output objects from GeosBlockExtractor filter and merge internal blocks volumeBlockExtracted: vtkMultiBlockDataSet = blockExtractor.extractedGeosDomain.volume - outputCells.ShallowCopy( mergeBlocksFilter( volumeBlockExtracted, False ) ) + outputCells.ShallowCopy( mergeBlocksFilter( volumeBlockExtracted, False, "Volume" ) ) outputCells.Modified() if extractFault: faultBlockExtracted: vtkMultiBlockDataSet = blockExtractor.extractedGeosDomain.fault - outputFaults.ShallowCopy( mergeBlocksFilter( faultBlockExtracted, True ) ) + outputFaults.ShallowCopy( mergeBlocksFilter( faultBlockExtracted, True, "Fault" ) ) outputFaults.Modified() if extractWell: wellBlockExtracted: vtkMultiBlockDataSet = blockExtractor.extractedGeosDomain.well - outputWells.ShallowCopy( mergeBlocksFilter( wellBlockExtracted, False ) ) + outputWells.ShallowCopy( mergeBlocksFilter( wellBlockExtracted, False, "Well" ) ) outputWells.Modified() return -def mergeBlocksFilter( mesh: vtkMultiBlockDataSet, - convertSurfaces: bool = False, ) -> vtkMultiBlockDataSet: + +def mergeBlocksFilter( + mesh: vtkMultiBlockDataSet, + convertSurfaces: bool = False, + domainToMerge: str = "Volume", +) -> vtkMultiBlockDataSet: """Apply vtk merge block filter on input multi block mesh. Args: mesh (vtkMultiBlockDataSet): Mesh to merge. convertSurfaces (bool, optional): True to convert surface from vtkUnstructuredGrid to vtkPolyData. Defaults to False. + domainToMerge (str, optional): The name of the GEOS domain processed. + Defaults to "Volume". Returns: vtkMultiBlockDataSet: Mesh composed of internal merged blocks. """ - mergeBlockFilter: GeosBlockMerge = GeosBlockMerge( mesh, convertSurfaces, True ) + loggerName = f"GEOS Block Merge for the domain { domainToMerge }." + mergeBlockFilter: GeosBlockMerge = GeosBlockMerge( mesh, convertSurfaces, True, loggerName ) if not mergeBlockFilter.logger.hasHandlers(): mergeBlockFilter.setLoggerHandler( VTKHandler() ) mergeBlockFilter.applyFilter() diff --git a/geos-utils/src/geos/utils/Logger.py b/geos-utils/src/geos/utils/Logger.py index 473f98f2..099ce907 100644 --- a/geos-utils/src/geos/utils/Logger.py +++ b/geos-utils/src/geos/utils/Logger.py @@ -275,7 +275,7 @@ def getLogger( title: str, use_color: bool = False ) -> Logger: """ logger = logging.getLogger( title ) # Only configure the logger (add handlers, set level) if it hasn't been configured before. - if not logger.hasHandlers(): # More Pythonic way to check if logger.handlers is empty + if len( logger.handlers ) == 0: logger.setLevel( INFO ) # Set the desired default level for this logger # Create and add the stream handler ch = logging.StreamHandler()