diff --git a/docs/geos_pv_docs/pythonViewUtils.rst b/docs/geos_pv_docs/pythonViewUtils.rst new file mode 100644 index 00000000..22e29c9e --- /dev/null +++ b/docs/geos_pv_docs/pythonViewUtils.rst @@ -0,0 +1,22 @@ +PythonViewUtils Package +============================ + +This package includes utilities to display cross-plot using the Python View from Paraview. + + +geos.pv.pythonViewUtils.Figure2DGenerator module +---------------------------------------------------------- + +.. automodule:: geos.pv.pythonViewUtils.Figure2DGenerator + :members: + :undoc-members: + :show-inheritance: + +geos.pv.pythonViewUtils.functionsFigure2DGenerator module +------------------------------------------------------------------- + +.. automodule:: geos.pv.pythonViewUtils.functionsFigure2DGenerator + :members: + :undoc-members: + :show-inheritance: + diff --git a/docs/geos_pv_docs/utilities.rst b/docs/geos_pv_docs/utilities.rst index 9b026137..052d49e6 100644 --- a/docs/geos_pv_docs/utilities.rst +++ b/docs/geos_pv_docs/utilities.rst @@ -8,4 +8,6 @@ Utilities pyplotUtils + pythonViewUtils + utils diff --git a/geos-pv/src/geos/pv/plugins/PVPythonViewConfigurator.py b/geos-pv/src/geos/pv/plugins/PVPythonViewConfigurator.py new file mode 100755 index 00000000..5ba12bd1 --- /dev/null +++ b/geos-pv/src/geos/pv/plugins/PVPythonViewConfigurator.py @@ -0,0 +1,858 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright 2023-2024 TotalEnergies. +# SPDX-FileContributor: Alexandre Benedicto, Martin Lemay +# ruff: noqa: E402 # disable Module level import not at top of file +import sys +from pathlib import Path +from typing import Any, Union, cast + +import pandas as pd # type: ignore[import-untyped] +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() + +import geos.pv.utils.paraviewTreatments as pvt +from geos.pv.utils.checkboxFunction import ( # type: ignore[attr-defined] + createModifiedCallback, ) +from geos.pv.utils.DisplayOrganizationParaview import ( + DisplayOrganizationParaview, ) +from geos.pv.pyplotUtils.matplotlibOptions import ( + FontStyleEnum, + FontWeightEnum, + LegendLocationEnum, + LineStyleEnum, + MarkerStyleEnum, + OptionSelectionEnum, + optionEnumToXml, +) +from paraview.simple import ( # type: ignore[import-not-found] + GetActiveSource, GetActiveView, Render, Show, servermanager, +) +from paraview.util.vtkAlgorithm import ( # type: ignore[import-not-found] + VTKPythonAlgorithmBase, smdomain, smhint, smproperty, smproxy, +) +from vtkmodules.vtkCommonCore import ( + vtkDataArraySelection, + vtkInformation, + vtkInformationVector, +) + +__doc__ = """ +PVPythonViewConfigurator is a Paraview plugin that allows to create cross-plots +from input data using the PythonView. + +Input type is vtkDataObject. + +This filter results in opening a new Python View window and displaying cross-plot. + +To use it: + +* Load the module in Paraview: Tools>Manage Plugins...>Load new>PVPythonViewConfigurator. +* Select the vtkDataObject containing the data to plot. +* Search and Apply PVPythonViewConfigurator Filter. + +""" + + +@smproxy.filter( name="PVPythonViewConfigurator", label="Python View Configurator" ) +@smhint.xml( '' ) +@smproperty.input( name="Input" ) +@smdomain.datatype( dataTypes=[ "vtkDataObject" ], composite_data_supported=True ) +class PVPythonViewConfigurator( VTKPythonAlgorithmBase ): + + def __init__( self: Self ) -> None: + """Paraview plugin to create cross-plots in a Python View. + + Input is a vtkDataObject. + """ + super().__init__( nInputPorts=1, nOutputPorts=1 ) + # Python view layout and object. + self.m_layoutName: str = "" + self.m_pythonView: Any + self.m_organizationDisplay = DisplayOrganizationParaview() + self.buildNewLayoutWithPythonView() + + # Input source and curve names. + inputSource = GetActiveSource() + dataset = servermanager.Fetch( inputSource ) + dataframe: pd.DataFrame = pvt.vtkToDataframe( dataset ) + self.m_pathPythonViewScript: Path = geos_pv_path / "src/geos/pv/pythonViewUtils/mainPythonView.py" + + # Checkboxes. + self.m_modifyInputs: int = 1 + self.m_modifyCurves: int = 1 + self.m_multiplyCurves: int = 0 + + # Checkboxes curves available from the data of pipeline. + self.m_validSources = vtkDataArraySelection() + self.m_curvesToPlot = vtkDataArraySelection() + self.m_curvesMinus1 = vtkDataArraySelection() + self.m_validSources.AddObserver( "ModifiedEvent", createModifiedCallback( self ) ) # type: ignore[arg-type] + self.m_curvesToPlot.AddObserver( "ModifiedEvent", createModifiedCallback( self ) ) # type: ignore[arg-type] + self.m_curvesMinus1.AddObserver( "ModifiedEvent", createModifiedCallback( self ) ) # type: ignore[arg-type] + validSourceNames: set[ str ] = pvt.getPossibleSourceNames() + for sourceName in validSourceNames: + self.m_validSources.AddArray( sourceName ) + validColumnsDataframe: list[ str ] = list( dataframe.columns ) + for name in list( dataframe.columns ): + for axis in [ "X", "Y", "Z" ]: + if "Points" + axis in name and "Points" + axis + "__" in name: + doublePosition: int = validColumnsDataframe.index( "Points" + axis ) + validColumnsDataframe.pop( doublePosition ) + break + self.m_validColumnsDataframe: list[ str ] = sorted( validColumnsDataframe, key=lambda x: x.lower() ) + for curveName in validColumnsDataframe: + self.m_curvesToPlot.AddArray( curveName ) + self.m_curvesMinus1.AddArray( curveName ) + self.m_validSources.DisableAllArrays() + self.m_curvesToPlot.DisableAllArrays() + self.m_curvesMinus1.DisableAllArrays() + self.m_curveToUse: str = "" + # To change the aspects of curves. + self.m_curvesToModify: set[ str ] = pvt.integrateSourceNames( validSourceNames, set( validColumnsDataframe ) ) + self.m_color: tuple[ float, float, float ] = ( 0.0, 0.0, 0.0 ) + self.m_lineStyle: str = LineStyleEnum.SOLID.optionValue + self.m_lineWidth: float = 1.0 + self.m_markerStyle: str = MarkerStyleEnum.NONE.optionValue + self.m_markerSize: float = 1.0 + + # User choices. + self.m_userChoices: dict[ str, Any ] = { + "variableName": "", + "curveNames": [], + "curveConvention": [], + "inputNames": [], + "plotRegions": False, + "reverseXY": False, + "logScaleX": False, + "logScaleY": False, + "minorticks": False, + "displayTitle": True, + "title": "title1", + "titleStyle": FontStyleEnum.NORMAL.optionValue, + "titleWeight": FontWeightEnum.BOLD.optionValue, + "titleSize": 12, + "legendDisplay": True, + "legendPosition": LegendLocationEnum.BEST.optionValue, + "legendSize": 10, + "removeJobName": True, + "removeRegions": False, + "curvesAspect": {}, + } + + def getUserChoices( self: Self ) -> dict[ str, Any ]: + """Access the m_userChoices attribute. + + Returns: + dict[str] : The user choices for the figure. + """ + return self.m_userChoices + + def getInputNames( self: Self ) -> set[ str ]: + """Get source names from user selection. + + Returns: + set[str] : Source names from ParaView pipeline. + """ + inputAvailable = self.a01GetInputSources() + inputNames: set[ str ] = set( pvt.getArrayChoices( inputAvailable ) ) + return inputNames + + def defineInputNames( self: Self ) -> None: + """Adds the input names to the userChoices.""" + inputNames: set[ str ] = self.getInputNames() + self.m_userChoices[ "inputNames" ] = inputNames + + def defineUserChoicesCurves( self: Self ) -> None: + """Define user choices for curves to plot.""" + sourceNames: set[ str ] = self.getInputNames() + dasPlot = self.b02GetCurvesToPlot() + dasMinus1 = self.b07GetCurveConvention() + curveNames: set[ str ] = set( pvt.getArrayChoices( dasPlot ) ) + minus1Names: set[ str ] = set( pvt.getArrayChoices( dasMinus1 ) ) + toUse1: set[ str ] = pvt.integrateSourceNames( sourceNames, curveNames ) + toUse2: set[ str ] = pvt.integrateSourceNames( sourceNames, minus1Names ) + self.m_userChoices[ "curveNames" ] = tuple( toUse1 ) + self.m_userChoices[ "curveConvention" ] = tuple( toUse2 ) + + def defineCurvesAspect( self: Self ) -> None: + """Define user choices for curve aspect properties.""" + curveAspect: tuple[ tuple[ float, float, float ], str, float, str, float ] = ( self.getCurveAspect() ) + curveName: str = self.getCurveToUse() + self.m_userChoices[ "curvesAspect" ][ curveName ] = curveAspect + + def buildPythonViewScript( self: Self ) -> str: + """Builds the Python script used to launch the Python View. + + The script is returned as a string to be then injected in the Python + View. + + Returns: + str: Complete Python View script. + """ + sourceNames: set[ str ] = self.getInputNames() + userChoices: dict[ str, Any ] = self.getUserChoices() + script: str = f"timestep = '{str(GetActiveView().ViewTime)}'\n" + script += f"sourceNames = {sourceNames}\n" + script += f"variableName = '{userChoices['variableName']}'\n" + script += f"dir_path = '{geos_pv_path}'\n" + script += f"userChoices = {userChoices}\n\n\n" + with self.m_pathPythonViewScript.open() as file: + fileContents = file.read() + script += fileContents + return script + + def buildNewLayoutWithPythonView( self: Self ) -> None: + """Create a new Python View layout.""" + # We first built the new layout. + layout_names: list[ str ] = self.m_organizationDisplay.getLayoutsNames() + nb_layouts: int = len( layout_names ) + # Imagine two layouts already exists, the new one will be named "Layout #3". + layoutName: str = "Layout #" + str( nb_layouts + 1 ) + # Check that we that the layoutName is new and does not belong to the list of layout_names, + # if not we modify the layoutName until it is a new one. + if layoutName in layout_names: + cpt: int = 2 + while layoutName in layout_names: + layoutName = "Layout #" + str( nb_layouts + cpt ) + cpt += 1 + self.m_organizationDisplay.addLayout( layoutName ) + self.m_layoutName = layoutName + + # We then build the new python view. + self.m_organizationDisplay.addViewToLayout( "PythonView", layoutName, 0 ) + self.m_pythonView = self.m_organizationDisplay.getLayoutViews()[ layoutName ][ 0 ] + Show( GetActiveSource(), self.m_pythonView, "PythonRepresentation" ) + + # Widgets definition + """The names of the @smproperty methods command names below have a letter in lower case in + front because PARAVIEW displays properties in the alphabetical order. + See https://gitlab.kitware.com/paraview/paraview/-/issues/21493 for possible improvements on + this issue.""" + + @smproperty.dataarrayselection( name="InputSources" ) + def a01GetInputSources( self: Self ) -> vtkDataArraySelection: + """Get all valid sources for the filter. + + Returns: + vtkDataArraySelection: Valid data sources. + """ + return self.m_validSources + + @smproperty.xml( """ + + """ ) + def a02GroupFlow( self: Self ) -> None: + """Organize groups.""" + self.Modified() + + @smproperty.stringvector( name="CurvesAvailable", information_only="1" ) + def b00GetCurvesAvailable( self: Self ) -> list[ str ]: + """Get the available curves. + + Returns: + list[str]: List of curves. + """ + return self.m_validColumnsDataframe + + @smproperty.stringvector( name="Abscissa", number_of_elements="1" ) + @smdomain.xml( """ + + """ ) + def b01SetVariableName( self: Self, name: str ) -> None: + """Set the name of X axis variable. + + Args: + name (str): Name of the variable. + """ + self.m_userChoices[ "variableName" ] = name + self.Modified() + + @smproperty.dataarrayselection( name="Ordinate" ) + def b02GetCurvesToPlot( self: Self ) -> vtkDataArraySelection: + """Get the curves to plot. + + Returns: + vtkDataArraySelection: Data to plot. + """ + return self.m_curvesToPlot + + @smproperty.intvector( name="PlotsPerRegion", label="PlotsPerRegion", default_values=0 ) + @smdomain.xml( """""" ) + def b03SetPlotsPerRegion( self: Self, boolean: bool ) -> None: + """Set plot per region option. + + Args: + boolean (bool): User choice. + """ + self.m_userChoices[ "plotRegions" ] = boolean + self.Modified() + + @smproperty.xml( """ + + + + """ ) + def b04GroupFlow( self: Self ) -> None: + """Organized groups.""" + self.Modified() + + @smproperty.intvector( + name="CurveConvention", + label="Select Curves To Change Convention", + default_values=0, + ) + @smdomain.xml( """""" ) + def b05SetCurveConvention( self: Self, boolean: bool ) -> None: + """Select Curves To Change Convention. + + Args: + boolean (bool): User choice. + """ + self.m_multiplyCurves = boolean + + @smproperty.xml( """ + + """ ) + def b06GroupFlow( self: Self ) -> None: + """Organized groups.""" + self.Modified() + + @smproperty.dataarrayselection( name="CurveConventionSelection" ) + def b07GetCurveConvention( self: Self ) -> vtkDataArraySelection: + """Get the curves to change convention. + + Returns: + vtkDataArraySelection: Selected curves to change convention. + """ + return self.m_curvesMinus1 + + @smproperty.xml( """ + + + """ ) + def b08GroupFlow( self: Self ) -> None: + """Organized groups.""" + self.Modified() + + @smproperty.intvector( name="EditAxisProperties", label="Edit Axis Properties", default_values=0 ) + @smdomain.xml( """""" ) + def c01SetEditAxisProperties( self: Self, boolean: bool ) -> None: + """Set option to edit axis properties. + + Args: + boolean (bool): User choice. + """ + self.Modified() + + @smproperty.xml( """ + + """ ) + def c02GroupFlow( self: Self ) -> None: + """Organized groups.""" + self.Modified() + + @smproperty.intvector( name="ReverseXY", label="Reverse XY Axes", default_values=0 ) + @smdomain.xml( """""" ) + def c02SetReverseXY( self: Self, boolean: bool ) -> None: + """Set option to reverse X and Y axes. + + Args: + boolean (bool): User choice. + """ + self.m_userChoices[ "reverseXY" ] = boolean + self.Modified() + + @smproperty.intvector( name="LogScaleX", label="X Axis Log Scale", default_values=0 ) + @smdomain.xml( """""" ) + def c03SetReverseXY( self: Self, boolean: bool ) -> None: + """Set option to log scale for X axis. + + Args: + boolean (bool): User choice. + """ + self.m_userChoices[ "logScaleX" ] = boolean + self.Modified() + + @smproperty.intvector( name="LogScaleY", label="Y Axis Log Scale", default_values=0 ) + @smdomain.xml( """""" ) + def c04SetReverseXY( self: Self, boolean: bool ) -> None: + """Set option to log scale for Y axis. + + Args: + boolean (bool): user choice. + """ + self.m_userChoices[ "logScaleY" ] = boolean + self.Modified() + + @smproperty.intvector( name="Minorticks", label="Display Minor ticks", default_values=0 ) + @smdomain.xml( """""" ) + def c05SetMinorticks( self: Self, boolean: bool ) -> None: + """Set option to display minor ticks. + + Args: + boolean (bool): User choice. + """ + self.m_userChoices[ "minorticks" ] = boolean + self.Modified() + + @smproperty.intvector( name="CustomAxisLim", label="Use Custom Axis Limits", default_values=0 ) + @smdomain.xml( """""" ) + def c06SetCustomAxisLim( self: Self, boolean: bool ) -> None: + """Set option to define axis limits. + + Args: + boolean (bool): User choice. + """ + self.m_userChoices[ "customAxisLim" ] = boolean + self.Modified() + + @smproperty.doublevector( name="LimMinX", label="X min", default_values=-1e36 ) + def c07LimMinX( self: Self, value: float ) -> None: + """Set X axis min. + + Args: + value (float): X axis min. + """ + value2: Union[ float, None ] = value + if value2 == -1e36: + value2 = None + self.m_userChoices[ "limMinX" ] = value2 + self.Modified() + + @smproperty.doublevector( name="LimMaxX", label="X max", default_values=1e36 ) + def c08LimMaxX( self: Self, value: float ) -> None: + """Set X axis max. + + Args: + value (float): X axis max. + """ + value2: Union[ float, None ] = value + if value2 == 1e36: + value2 = None + self.m_userChoices[ "limMaxX" ] = value2 + self.Modified() + + @smproperty.doublevector( name="LimMinY", label="Y min", default_values=-1e36 ) + def c09LimMinY( self: Self, value: float ) -> None: + """Set Y axis min. + + Args: + value (float): Y axis min. + """ + value2: Union[ float, None ] = value + if value2 == -1e36: + value2 = None + self.m_userChoices[ "limMinY" ] = value2 + self.Modified() + + @smproperty.doublevector( name="LimMaxY", label="Y max", default_values=1e36 ) + def c10LimMaxY( self: Self, value: float ) -> None: + """Set Y axis max. + + Args: + value (float): Y axis max. + """ + value2: Union[ float, None ] = value + if value2 == 1e36: + value2 = None + self.m_userChoices[ "limMaxY" ] = value2 + self.Modified() + + @smproperty.xml( """ + + + + + + """ ) + def c11GroupFlow( self: Self ) -> None: + """Organized groups.""" + self.Modified() + + @smproperty.xml( """ + + + + + + + """ ) + def c12GroupFlow( self: Self ) -> None: + """Organized groups.""" + self.Modified() + + @smproperty.intvector( name="DisplayTitle", label="Display Title", default_values=1 ) + @smdomain.xml( """""" ) + def d01SetDisplayTitle( self: Self, boolean: bool ) -> None: + """Set option to display title. + + Args: + boolean (bool): User choice. + """ + self.m_userChoices[ "displayTitle" ] = boolean + self.Modified() + + @smproperty.xml( """ + + """ ) + def d02GroupFlow( self: Self ) -> None: + """Organized groups.""" + self.Modified() + + @smproperty.stringvector( name="Title", default_values="title1" ) + def d03SetTitlePlot( self: Self, title: str ) -> None: + """Set title. + + Args: + title (str): Title. + """ + self.m_userChoices[ "title" ] = title + self.Modified() + + @smproperty.intvector( name="TitleStyle", label="Title Style", default_values=0 ) + @smdomain.xml( optionEnumToXml( cast( OptionSelectionEnum, FontStyleEnum ) ) ) + def d04SetTitleStyle( self: Self, value: int ) -> None: + """Set title font style. + + Args: + value (int): Title font style index in FontStyleEnum. + """ + choice = list( FontStyleEnum )[ value ] + self.m_userChoices[ "titleStyle" ] = choice.optionValue + self.Modified() + + @smproperty.intvector( name="TitleWeight", label="Title Weight", default_values=1 ) + @smdomain.xml( optionEnumToXml( cast( OptionSelectionEnum, FontWeightEnum ) ) ) + def d05SetTitleWeight( self: Self, value: int ) -> None: + """Set title font weight. + + Args: + value (int): Title font weight index in FontWeightEnum. + """ + choice = list( FontWeightEnum )[ value ] + self.m_userChoices[ "titleWeight" ] = choice.optionValue + self.Modified() + + @smproperty.intvector( name="TitleSize", label="Title Size", default_values=12 ) + @smdomain.xml( """""" ) + def d06SetTitleSize( self: Self, size: float ) -> None: + """Set title font size. + + Args: + size (float): Title font size between 1 and 50. + """ + self.m_userChoices[ "titleSize" ] = size + self.Modified() + + @smproperty.xml( """ + panel_visibility="advanced"> + + + + + + """ ) + def d07PropertyGroup( self: Self ) -> None: + """Organized groups.""" + self.Modified() + + @smproperty.intvector( name="DisplayLegend", label="Display Legend", default_values=1 ) + @smdomain.xml( """""" ) + def e00SetDisplayLegend( self: Self, boolean: bool ) -> None: + """Set option to display legend. + + Args: + boolean (bool): User choice. + """ + self.m_userChoices[ "displayLegend" ] = boolean + self.Modified() + + @smproperty.xml( """ + + """ ) + def e01PropertyGroup( self: Self ) -> None: + """Organized groups.""" + self.Modified() + + @smproperty.intvector( name="LegendPosition", label="Legend Position", default_values=0 ) + @smdomain.xml( optionEnumToXml( cast( OptionSelectionEnum, LegendLocationEnum ) ) ) + def e02SetLegendPosition( self: Self, value: int ) -> None: + """Set legend position. + + Args: + value (int): Legend position index in LegendLocationEnum. + """ + choice = list( LegendLocationEnum )[ value ] + self.m_userChoices[ "legendPosition" ] = choice.optionValue + self.Modified() + + @smproperty.intvector( name="LegendSize", label="Legend Size", default_values=10 ) + @smdomain.xml( """""" ) + def e03SetLegendSize( self: Self, size: float ) -> None: + """Set legend font size. + + Args: + size (float): Legend font size between 1 and 50. + """ + self.m_userChoices[ "legendSize" ] = size + self.Modified() + + @smproperty.intvector( name="RemoveJobName", label="Remove Job Name in legend", default_values=1 ) + @smdomain.xml( """""" ) + def e04SetRemoveJobName( self: Self, boolean: bool ) -> None: + """Set option to remove job names from legend. + + Args: + boolean (bool): User choice. + """ + self.m_userChoices[ "removeJobName" ] = boolean + self.Modified() + + @smproperty.intvector( + name="RemoveRegionsName", + label="Remove Regions Name in legend", + default_values=0, + ) + @smdomain.xml( """""" ) + def e05SetRemoveRegionsName( self: Self, boolean: bool ) -> None: + """Set option to remove region names from legend. + + Args: + boolean (bool): User choice. + """ + self.m_userChoices[ "removeRegions" ] = boolean + self.Modified() + + @smproperty.xml( """ + + + + + + """ ) + def e06PropertyGroup( self: Self ) -> None: + """Organized groups.""" + self.Modified() + + @smproperty.intvector( name="ModifyCurvesAspect", label="Edit Curve Graphics", default_values=1 ) + @smdomain.xml( """""" ) + def f01SetModifyCurvesAspect( self: Self, boolean: bool ) -> None: + """Set option to change curve aspects. + + Args: + boolean (bool): User choice. + """ + self.m_modifyCurvesAspect = boolean + + @smproperty.xml( """ + + """ ) + def f02PropertyGroup( self: Self ) -> None: + """Organized groups.""" + self.Modified() + + @smproperty.stringvector( name="CurvesInfo", information_only="1" ) + def f03GetCurveNames( self: Self ) -> list[ str ]: + """Get curves to modify aspects. + + Returns: + set[str]: Curves to modify aspects. + """ + return list( self.m_curvesToModify ) + + @smproperty.stringvector( name="CurveToModify", number_of_elements="1" ) + @smdomain.xml( """ + + """ ) + def f04SetCircleID( self: Self, value: str ) -> None: + """Set m_curveToUse. + + Args: + value (float): Value of m_curveToUse. + """ + self.m_curveToUse = value + self.Modified() + + def getCurveToUse( self: Self ) -> str: + """Get m_curveToUse.""" + return self.m_curveToUse + + @smproperty.intvector( name="LineStyle", label="Line Style", default_values=1 ) + @smdomain.xml( optionEnumToXml( cast( OptionSelectionEnum, LineStyleEnum ) ) ) + def f05SetLineStyle( self: Self, value: int ) -> None: + """Set line style. + + Args: + value (int): Line style index in LineStyleEnum. + """ + choice = list( LineStyleEnum )[ value ] + self.m_lineStyle = choice.optionValue + self.Modified() + + @smproperty.doublevector( name="LineWidth", default_values=1.0 ) + @smdomain.xml( """""" ) + def f06SetLineWidth( self: Self, value: float ) -> None: + """Set line width. + + Args: + value (float): Line width between 1 and 10. + """ + self.m_lineWidth = value + self.Modified() + + @smproperty.intvector( name="MarkerStyle", label="Marker Style", default_values=0 ) + @smdomain.xml( optionEnumToXml( cast( LegendLocationEnum, MarkerStyleEnum ) ) ) + def f07SetMarkerStyle( self: Self, value: int ) -> None: + """Set marker style. + + Args: + value (int): Marker style index in MarkerStyleEnum. + """ + choice = list( MarkerStyleEnum )[ value ] + self.m_markerStyle = choice.optionValue + self.Modified() + + @smproperty.doublevector( name="MarkerSize", default_values=1.0 ) + @smdomain.xml( """""" ) + def f08SetMarkerSize( self: Self, value: float ) -> None: + """Set marker size. + + Args: + value (float): Size of markers between 1 and 30. + """ + self.m_markerSize = value + self.Modified() + + @smproperty.xml( """ + + + + + + + + """ ) + def f09PropertyGroup( self: Self ) -> None: + """Organized groups.""" + self.Modified() + + @smproperty.doublevector( name="ColorEnvelop", default_values=[ 0, 0, 0 ], number_of_elements=3 ) + @smdomain.xml( """""" ) + def f10SetColor( self: Self, value0: float, value1: float, value2: float ) -> None: + """Set envelope color. + + Args: + value0 (float): Red color between 0 and 1. + value1 (float): Green color between 0 and 1. + value2 (float): Blue color between 0 and 1. + """ + self.m_color = ( value0, value1, value2 ) + self.Modified() + + @smproperty.xml( """ + + + """ ) + def f11PropertyGroup( self: Self ) -> None: + """Organized groups.""" + self.Modified() + + def getCurveAspect( self: Self, ) -> tuple[ tuple[ float, float, float ], str, float, str, float ]: + """Get curve aspect properties according to user choices. + + Returns: + tuple: (color, lineStyle, linewidth, marker, markerSize) + """ + return ( + self.m_color, + self.m_lineStyle, + self.m_lineWidth, + self.m_markerStyle, + self.m_markerSize, + ) + + 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(), "vtkDataObject" ) + else: + info.Set( self.INPUT_REQUIRED_DATA_TYPE(), "vtkDataObject" ) + return 1 + + def RequestDataObject( + self: Self, + request: vtkInformation, + inInfoVec: list[ vtkInformationVector ], + outInfoVec: vtkInformationVector, + ) -> int: + """Inherited from VTKPythonAlgorithmBase::RequestDataObject. + + Args: + request (vtkInformation): Request. + inInfoVec (list[vtkInformationVector]): Input objects. + outInfoVec (vtkInformationVector): Output objects. + + Returns: + int: 1 if calculation successfully ended, 0 otherwise. + """ + inData = self.GetInputData( inInfoVec, 0, 0 ) + outData = self.GetOutputData( outInfoVec, 0 ) + assert inData is not None + if outData is None or ( not outData.IsA( inData.GetClassName() ) ): + outData = inData.NewInstance() + outInfoVec.GetInformationObject( 0 ).Set( outData.DATA_OBJECT(), outData ) + return super().RequestDataObject( request, inInfoVec, outInfoVec ) # type: ignore[no-any-return] + + def RequestData( + self: Self, + request: vtkInformation, # noqa: F841 + inInfoVec: list[ vtkInformationVector ], # noqa: F841 + outInfoVec: vtkInformationVector, # noqa: F841 + ) -> 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. + """ + # pythonViewGeneration + assert self.m_pythonView is not None, "No Python View was found." + viewSize = GetActiveView().ViewSize + self.m_userChoices[ "ratio" ] = viewSize[ 0 ] / viewSize[ 1 ] + self.defineInputNames() + self.defineUserChoicesCurves() + self.defineCurvesAspect() + self.m_pythonView.Script = self.buildPythonViewScript() + Render() + return 1 diff --git a/geos-pv/src/geos/pv/pythonViewUtils/Figure2DGenerator.py b/geos-pv/src/geos/pv/pythonViewUtils/Figure2DGenerator.py new file mode 100755 index 00000000..00eceded --- /dev/null +++ b/geos-pv/src/geos/pv/pythonViewUtils/Figure2DGenerator.py @@ -0,0 +1,137 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright 2023-2024 TotalEnergies. +# SPDX-FileContributor: Alexandre Benedicto +from logging import Logger +from typing import Any + +import pandas as pd # type: ignore[import-untyped] +from matplotlib import axes, figure, lines # type: ignore[import-untyped] +from matplotlib.font_manager import ( # type: ignore[import-untyped] + FontProperties, # type: ignore[import-untyped] +) +from typing_extensions import Self + +import geos.pv.pythonViewUtils.functionsFigure2DGenerator as fcts + + +class Figure2DGenerator: + + def __init__( self: Self, dataframe: pd.DataFrame, userChoices: dict[ str, list[ str ] ], logger: Logger ) -> None: + """Utility to create cross plots using Python View. + + We want to plot f(X) = Y where in this class, + "X" will be called "variable", "Y" will be called "curves". + + Args: + dataframe (pd.DataFrame): Data to plot. + userChoices (dict[str, list[str]]): User choices. + logger (Logger): Logger to use. + """ + self.m_dataframe: pd.DataFrame = dataframe + self.m_userChoices: dict[ str, Any ] = userChoices + self.m_fig: figure.Figure + self.m_axes: list[ axes._axes.Axes ] = [] + self.m_lines: list[ lines.Line2D ] = [] + self.m_labels: list[ str ] = [] + self.m_logger: Logger = logger + + try: + # Apply minus 1 multiplication on certain columns. + self.initMinus1Multiplication() + # Defines m_fig, m_axes, m_lines and m_labels. + self.plotInitialFigure() + # Then to edit and customize the figure. + self.enhanceFigure() + self.m_logger.info( "Data were successfully plotted." ) + + except Exception as e: + mess: str = "Plot creation failed due to:" + self.m_logger.critical( mess ) + self.m_logger.critical( e, exc_info=True ) + + def initMinus1Multiplication( self: Self ) -> None: + """Multiply by -1 certain columns of the input dataframe.""" + df: pd.DataFrame = self.m_dataframe.copy( deep=True ) + minus1CurveNames: list[ str ] = self.m_userChoices[ "curveConvention" ] + for name in minus1CurveNames: + df[ name ] = df[ name ] * ( -1 ) + self.m_dataframe = df + + def enhanceFigure( self: Self ) -> None: + """Apply all the enhancement features to the initial figure.""" + self.changeTitle() + self.changeMinorticks() + self.changeAxisScale() + self.changeAxisLimits() + + def plotInitialFigure( self: Self ) -> None: + """Generates a figure and axes objects from matplotlib. + + The figure plots all the curves along the X or Y axis, with legend and + label for X and Y. + """ + if self.m_userChoices[ "plotRegions" ]: + if not self.m_userChoices[ "reverseXY" ]: + ( fig, ax_all, lines, labels ) = fcts.multipleSubplots( self.m_dataframe, self.m_userChoices ) + else: + ( fig, ax_all, lines, labels ) = fcts.multipleSubplotsInverted( self.m_dataframe, self.m_userChoices ) + else: + if not self.m_userChoices[ "reverseXY" ]: + ( fig, ax_all, lines, labels ) = fcts.oneSubplot( self.m_dataframe, self.m_userChoices ) + else: + ( fig, ax_all, lines, labels ) = fcts.oneSubplotInverted( self.m_dataframe, self.m_userChoices ) + self.m_fig = fig + self.m_axes = ax_all + self.m_lines = lines + self.m_labels = labels + + def changeTitle( self: Self ) -> None: + """Update title of the first axis of the figure based on user choices.""" + if self.m_userChoices[ "displayTitle" ]: + title: str = self.m_userChoices[ "title" ] + fontTitle: FontProperties = fcts.buildFontTitle( self.m_userChoices ) + self.m_fig.suptitle( title, fontproperties=fontTitle ) + + def changeMinorticks( self: Self ) -> None: + """Set the minorticks on or off for every axes.""" + choice: bool = self.m_userChoices[ "minorticks" ] + if choice: + for ax in self.m_axes: + ax.minorticks_on() + else: + for ax in self.m_axes: + ax.minorticks_off() + + def changeAxisScale( self: Self ) -> None: + """Set the minorticks on or off for every axes.""" + for ax in self.m_axes: + if self.m_userChoices[ "logScaleX" ]: + ax.set_xscale( "log" ) + if self.m_userChoices[ "logScaleY" ]: + ax.set_yscale( "log" ) + + def changeAxisLimits( self: Self ) -> None: + """Update axis limits.""" + if self.m_userChoices[ "customAxisLim" ]: + for ax in self.m_axes: + xmin, xmax = ax.get_xlim() + if self.m_userChoices[ "limMinX" ] is not None: + xmin = self.m_userChoices[ "limMinX" ] + if self.m_userChoices[ "limMaxX" ] is not None: + xmax = self.m_userChoices[ "limMaxX" ] + ax.set_xlim( xmin, xmax ) + + ymin, ymax = ax.get_ylim() + if self.m_userChoices[ "limMinY" ] is not None: + ymin = self.m_userChoices[ "limMinY" ] + if self.m_userChoices[ "limMaxY" ] is not None: + ymax = self.m_userChoices[ "limMaxY" ] + ax.set_ylim( ymin, ymax ) + + def getFigure( self: Self ) -> figure.Figure: + """access the m_fig attribute. + + Returns: + figure.Figure: Figure containing all the plots. + """ + return self.m_fig diff --git a/geos-pv/src/geos/pv/pythonViewUtils/__init__.py b/geos-pv/src/geos/pv/pythonViewUtils/__init__.py new file mode 100755 index 00000000..e69de29b diff --git a/geos-pv/src/geos/pv/pythonViewUtils/functionsFigure2DGenerator.py b/geos-pv/src/geos/pv/pythonViewUtils/functionsFigure2DGenerator.py new file mode 100755 index 00000000..11395341 --- /dev/null +++ b/geos-pv/src/geos/pv/pythonViewUtils/functionsFigure2DGenerator.py @@ -0,0 +1,1376 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright 2023-2024 TotalEnergies. +# SPDX-FileContributor: Alexandre Benedicto +import math +from typing import Any + +import matplotlib.pyplot as plt # type: ignore[import-untyped] +import numpy as np +import numpy.typing as npt +import pandas as pd # type: ignore[import-untyped] +from matplotlib import axes, figure, lines # type: ignore[import-untyped] +from matplotlib.font_manager import ( # type: ignore[import-untyped] + FontProperties, # type: ignore[import-untyped] +) + +import geos.pv.geosLogReaderUtils.geosLogReaderFunctions as fcts +""" +Plotting tools for 2D figure and axes generation. +""" + + +def oneSubplot( + df: pd.DataFrame, + userChoices: dict[ str, Any ] ) -> tuple[ figure.Figure, list[ axes.Axes ], list[ lines.Line2D ], list[ str ] ]: + """Created a single subplot. + + From a dataframe, knowing which curves to plot along which variable, + generates a fig and its list of axes with the data plotted. + + Args: + df (pd.DataFrame): Dataframe containing at least two columns, + one named "variableName" and the other "curveName". + userChoices (dict[str, Any]): Choices made by widget selection + in PythonViewConfigurator filter. + + Returns: + tuple(figure.Figure, list[axes.Axes], + list[lines.Line2D] , list[str]): The fig and its list of axes. + """ + curveNames: list[ str ] = userChoices[ "curveNames" ] + variableName: str = userChoices[ "variableName" ] + curvesAspect: dict[ str, tuple[ tuple[ float, float, float ], str, float, str, + float ] ] = userChoices[ "curvesAspect" ] + associatedProperties: dict[ str, list[ str ] ] = associatePropertyToAxeType( curveNames ) + fig, ax = plt.subplots( constrained_layout=True ) + all_ax: list[ axes.Axes ] = setupAllAxes( ax, variableName, associatedProperties, True ) + lineList: list[ lines.Line2D ] = [] + labels: list[ str ] = [] + cpt_cmap: int = 0 + x: npt.NDArray[ np.float64 ] = df[ variableName ].to_numpy() + for cpt_ax, ( ax_name, propertyNames ) in enumerate( associatedProperties.items() ): + ax_to_use: axes.Axes = setupAxeToUse( all_ax, cpt_ax, ax_name, False ) + for propName in propertyNames: + y: npt.NDArray[ np.float64 ] = df[ propName ].to_numpy() + plotAxe( ax_to_use, x, y, propName, cpt_cmap, curvesAspect ) + cpt_cmap += 1 + new_lines, new_labels = ax_to_use.get_legend_handles_labels() + lineList += new_lines # type: ignore[arg-type] + labels += new_labels + labels, lineList = smartLabelsSorted( labels, lineList, userChoices ) + if userChoices[ "displayLegend" ]: + ax.legend( + lineList, + labels, + loc=userChoices[ "legendPosition" ], + fontsize=userChoices[ "legendSize" ], + ) + ax.grid() + return ( fig, all_ax, lineList, labels ) + + +def oneSubplotInverted( + df: pd.DataFrame, + userChoices: dict[ str, Any ] ) -> tuple[ figure.Figure, list[ axes.Axes ], list[ lines.Line2D ], list[ str ] ]: + """Created a single subplot with inverted X Y axes. + + From a dataframe, knowing which curves to plot along which variable, + generates a fig and its list of axes with the data plotted. + + Args: + df (pd.DataFrame): Dataframe containing at least two columns, + one named "variableName" and the other "curveName". + userChoices (dict[str, Any]): Choices made by widget selection + in PythonViewConfigurator filter. + + Returns: + tuple(figure.Figure, list[axes.Axes], + list[lines.Line2D] , list[str]): The fig and its list of axes. + """ + curveNames: list[ str ] = userChoices[ "curveNames" ] + variableName: str = userChoices[ "variableName" ] + curvesAspect: dict[ str, tuple[ tuple[ float, float, float ], str, float, str, + float ] ] = userChoices[ "curvesAspect" ] + associatedProperties: dict[ str, list[ str ] ] = associatePropertyToAxeType( curveNames ) + fig, ax = plt.subplots( constrained_layout=True ) + all_ax: list[ axes.Axes ] = setupAllAxes( ax, variableName, associatedProperties, False ) + linesList: list[ lines.Line2D ] = [] + labels: list[ str ] = [] + cpt_cmap: int = 0 + y: npt.NDArray[ np.float64 ] = df[ variableName ].to_numpy() + for cpt_ax, ( ax_name, propertyNames ) in enumerate( associatedProperties.items() ): + ax_to_use: axes.Axes = setupAxeToUse( all_ax, cpt_ax, ax_name, True ) + for propName in propertyNames: + x: npt.NDArray[ np.float64 ] = df[ propName ].to_numpy() + plotAxe( ax_to_use, x, y, propName, cpt_cmap, curvesAspect ) + cpt_cmap += 1 + new_lines, new_labels = ax_to_use.get_legend_handles_labels() + linesList += new_lines # type: ignore[arg-type] + labels += new_labels + labels, linesList = smartLabelsSorted( labels, linesList, userChoices ) + if userChoices[ "displayLegend" ]: + ax.legend( + linesList, + labels, + loc=userChoices[ "legendPosition" ], + fontsize=userChoices[ "legendSize" ], + ) + ax.grid() + return ( fig, all_ax, linesList, labels ) + + +def multipleSubplots( + df: pd.DataFrame, + userChoices: dict[ str, Any ] ) -> tuple[ figure.Figure, list[ axes.Axes ], list[ lines.Line2D ], list[ str ] ]: + """Created multiple subplots. + + From a dataframe, knowing which curves to plot along which variable, + generates a fig and its list of axes with the data plotted. + + Args: + df (pd.DataFrame): Dataframe containing at least two columns, + one named "variableName" and the other "curveName". + userChoices (dict[str, Any]): Choices made by widget selection + in PythonViewConfigurator filter. + + Returns: + tuple(figure.Figure, list[axes.Axes], + list[lines.Line2D] , list[str]): The fig and its list of axes. + """ + curveNames: list[ str ] = userChoices[ "curveNames" ] + variableName: str = userChoices[ "variableName" ] + curvesAspect: dict[ str, tuple[ tuple[ float, float, float ], str, float, str, + float ] ] = userChoices[ "curvesAspect" ] + ratio: float = userChoices[ "ratio" ] + assosIdentifiers: dict[ str, dict[ str, list[ str ] ] ] = associationIdentifiers( curveNames ) + nbr_suplots: int = len( assosIdentifiers.keys() ) + # if only one subplots needs to be created + if nbr_suplots == 1: + return oneSubplot( df, userChoices ) + + layout: tuple[ int, int, int ] = smartLayout( nbr_suplots, ratio ) + fig, axs0 = plt.subplots( layout[ 0 ], layout[ 1 ], constrained_layout=True ) + axs: list[ axes.Axes ] = axs0.flatten().tolist() # type: ignore[union-attr] + for i in range( layout[ 2 ] ): + fig.delaxes( axs[ -( i + 1 ) ] ) + all_lines: list[ lines.Line2D ] = [] + all_labels: list[ str ] = [] + # first loop for subplots + propertiesExtremas: dict[ str, tuple[ float, float ] ] = ( findExtremasPropertiesForAssociatedIdentifiers( + df, assosIdentifiers, True ) ) + for j, identifier in enumerate( assosIdentifiers.keys() ): + first_ax: axes.Axes = axs[ j ] + associatedProperties: dict[ str, list[ str ] ] = assosIdentifiers[ identifier ] + all_ax: list[ axes.Axes ] = setupAllAxes( first_ax, variableName, associatedProperties, True ) + axs += all_ax[ 1: ] + linesList: list[ lines.Line2D ] = [] + labels: list[ str ] = [] + cpt_cmap: int = 0 + x: npt.NDArray[ np.float64 ] = df[ variableName ].to_numpy() + # second loop for axes per subplot + for cpt_ax, ( ax_name, propertyNames ) in enumerate( associatedProperties.items() ): + ax_to_use: axes.Axes = setupAxeToUse( all_ax, cpt_ax, ax_name, False ) + for propName in propertyNames: + y: npt.NDArray[ np.float64 ] = df[ propName ].to_numpy() + plotAxe( ax_to_use, x, y, propName, cpt_cmap, curvesAspect ) + ax_to_use.set_ylim( *propertiesExtremas[ ax_name ] ) + cpt_cmap += 1 + new_lines, new_labels = ax_to_use.get_legend_handles_labels() + linesList += new_lines # type: ignore[arg-type] + all_lines += new_lines # type: ignore[arg-type] + labels += new_labels + all_labels += new_labels + labels, linesList = smartLabelsSorted( labels, linesList, userChoices ) + if userChoices[ "displayLegend" ]: + first_ax.legend( + linesList, + labels, + loc=userChoices[ "legendPosition" ], + fontsize=userChoices[ "legendSize" ], + ) + if userChoices[ "displayTitle" ]: + first_ax.set_title( identifier, fontsize=10 ) + first_ax.grid() + return ( fig, axs, all_lines, all_labels ) + + +def multipleSubplotsInverted( + df: pd.DataFrame, + userChoices: dict[ str, Any ] ) -> tuple[ figure.Figure, list[ axes.Axes ], list[ lines.Line2D ], list[ str ] ]: + """Created multiple subplots with inverted X Y axes. + + From a dataframe, knowing which curves to plot along which variable, + generates a fig and its list of axes with the data plotted. + + Args: + df (pd.DataFrame): Dataframe containing at least two columns, + one named "variableName" and the other "curveName". + userChoices (dict[str, Any]): Choices made by widget selection + in PythonViewConfigurator filter. + + Returns: + tuple(figure.Figure, list[axes.Axes], + list[lines.Line2D] , list[str]): The fig and its list of axes. + """ + curveNames: list[ str ] = userChoices[ "curveNames" ] + variableName: str = userChoices[ "variableName" ] + curvesAspect: dict[ str, tuple[ tuple[ float, float, float ], str, float, str, + float ] ] = userChoices[ "curvesAspect" ] + ratio: float = userChoices[ "ratio" ] + assosIdentifiers: dict[ str, dict[ str, list[ str ] ] ] = associationIdentifiers( curveNames ) + nbr_suplots: int = len( assosIdentifiers.keys() ) + # If only one subplots needs to be created. + if nbr_suplots == 1: + return oneSubplotInverted( df, userChoices ) + + layout: tuple[ int, int, int ] = smartLayout( nbr_suplots, ratio ) + fig, axs0 = plt.subplots( layout[ 0 ], layout[ 1 ], constrained_layout=True ) + axs: list[ axes.Axes ] = axs0.flatten().tolist() # type: ignore[union-attr] + for i in range( layout[ 2 ] ): + fig.delaxes( axs[ -( i + 1 ) ] ) + all_lines: list[ lines.Line2D ] = [] + all_labels: list[ str ] = [] + # First loop for subplots. + propertiesExtremas: dict[ str, tuple[ float, float ] ] = ( findExtremasPropertiesForAssociatedIdentifiers( + df, assosIdentifiers, True ) ) + for j, identifier in enumerate( assosIdentifiers.keys() ): + first_ax: axes.Axes = axs[ j ] + associatedProperties: dict[ str, list[ str ] ] = assosIdentifiers[ identifier ] + all_ax: list[ axes.Axes ] = setupAllAxes( first_ax, variableName, associatedProperties, False ) + axs += all_ax[ 1: ] + linesList: list[ lines.Line2D ] = [] + labels: list[ str ] = [] + cpt_cmap: int = 0 + y: npt.NDArray[ np.float64 ] = df[ variableName ].to_numpy() + # Second loop for axes per subplot. + for cpt_ax, ( ax_name, propertyNames ) in enumerate( associatedProperties.items() ): + ax_to_use: axes.Axes = setupAxeToUse( all_ax, cpt_ax, ax_name, True ) + for propName in propertyNames: + x: npt.NDArray[ np.float64 ] = df[ propName ].to_numpy() + plotAxe( ax_to_use, x, y, propName, cpt_cmap, curvesAspect ) + ax_to_use.set_xlim( propertiesExtremas[ ax_name ] ) + cpt_cmap += 1 + new_lines, new_labels = ax_to_use.get_legend_handles_labels() + linesList += new_lines # type: ignore[arg-type] + all_lines += new_lines # type: ignore[arg-type] + labels += new_labels + all_labels += new_labels + labels, linesList = smartLabelsSorted( labels, linesList, userChoices ) + if userChoices[ "displayLegend" ]: + first_ax.legend( + linesList, + labels, + loc=userChoices[ "legendPosition" ], + fontsize=userChoices[ "legendSize" ], + ) + if userChoices[ "displayTitle" ]: + first_ax.set_title( identifier, fontsize=10 ) + first_ax.grid() + return ( fig, axs, all_lines, all_labels ) + + +def setupAllAxes( + first_ax: axes.Axes, + variableName: str, + associatedProperties: dict[ str, list[ str ] ], + axisX: bool, +) -> list[ axes.Axes ]: + """Modify axis name and ticks with X or Y axis of all subplots. + + Args: + first_ax (axes.Axes): Subplot id. + variableName (str): Name of the axis. + associatedProperties (dict[str, list[str]]): Name of the properties. + axisX (bool): X (True) or Y (False) axis to modify. + + Returns: + list[axes.Axes]: Modified subplots. + """ + all_ax: list[ axes.Axes ] = [ first_ax ] + if axisX: + first_ax.set_xlabel( variableName ) + first_ax.ticklabel_format( style="sci", axis="x", scilimits=( 0, 0 ), useMathText=True ) + for i in range( 1, len( associatedProperties.keys() ) ): + second_ax = first_ax.twinx() + assert isinstance( second_ax, axes.Axes ) + all_ax.append( second_ax ) + all_ax[ i ].spines[ "right" ].set_position( ( "axes", 1 + 0.07 * ( i - 1 ) ) ) + all_ax[ i ].tick_params( axis="y", which="both", left=False, right=True ) + all_ax[ i ].yaxis.set_ticks_position( "right" ) + all_ax[ i ].yaxis.offsetText.set_position( ( 1.04 + 0.07 * ( i - 1 ), 0 ) ) + first_ax.yaxis.offsetText.set_position( ( -0.04, 0 ) ) + else: + first_ax.set_ylabel( variableName ) + first_ax.ticklabel_format( style="sci", axis="y", scilimits=( 0, 0 ), useMathText=True ) + for i in range( 1, len( associatedProperties.keys() ) ): + second_ax = first_ax.twiny() + assert isinstance( second_ax, axes.Axes ) + all_ax.append( second_ax ) + all_ax[ i ].spines[ "bottom" ].set_position( ( "axes", -0.08 * i ) ) + all_ax[ i ].xaxis.set_label_position( "bottom" ) + all_ax[ i ].tick_params( axis="x", which="both", bottom=True, top=False ) + all_ax[ i ].xaxis.set_ticks_position( "bottom" ) + return all_ax + + +def setupAxeToUse( all_ax: list[ axes.Axes ], axeId: int, ax_name: str, axisX: bool ) -> axes.Axes: + """Modify axis name and ticks with X or Y axis of subplot axeId in all_ax. + + Args: + all_ax (list[axes.Axes]): List of all subplots. + axeId (int): Id of the subplot. + ax_name (str): Name of the X or Y axis. + axisX (bool): X (True) or Y (False) axis to modify. + + Returns: + axes.Axes: Modified subplot. + """ + ax_to_use: axes.Axes = all_ax[ axeId ] + if axisX: + ax_to_use.set_xlabel( ax_name ) + ax_to_use.ticklabel_format( style="sci", axis="x", scilimits=( 0, 0 ), useMathText=True ) + else: + ax_to_use.set_ylabel( ax_name ) + ax_to_use.ticklabel_format( style="sci", axis="y", scilimits=( 0, 0 ), useMathText=True ) + return ax_to_use + + +def plotAxe( + ax_to_use: axes.Axes, + x: npt.NDArray[ np.float64 ], + y: npt.NDArray[ np.float64 ], + propertyName: str, + cpt_cmap: int, + curvesAspect: dict[ str, tuple[ tuple[ float, float, float ], str, float, str, float ] ], +) -> None: + """Plot x, y data using input ax_to_use according to curvesAspect. + + Args: + ax_to_use (axes.Axes): Subplot to use. + x (npt.NDArray[np.float64]): Abscissa data. + y (npt.NDArray[np.float64]): Ordinate data. + propertyName (str): Name of the property. + cpt_cmap (int): Colormap to use. + curvesAspect (dict[str, tuple[tuple[float, float, float],str, float, str, float]]): + User choices on curve aspect. + """ + cmap = plt.rcParams[ "axes.prop_cycle" ].by_key()[ "color" ][ cpt_cmap % 10 ] + mask = np.logical_and( np.isnan( x ), np.isnan( y ) ) + not_mask = ~mask + # Plot only when x and y values are not nan values. + if propertyName in curvesAspect: + asp: tuple[ tuple[ float, float, float ], str, float, str, float ] = curvesAspect[ propertyName ] + ax_to_use.plot( + x[ not_mask ], + y[ not_mask ], + label=propertyName, + color=asp[ 0 ], + linestyle=asp[ 1 ], + linewidth=asp[ 2 ], + marker=asp[ 3 ], + markersize=asp[ 4 ], + ) + else: + ax_to_use.plot( x[ not_mask ], y[ not_mask ], label=propertyName, color=cmap ) + + +def getExtremaAllAxes( axes: list[ axes.Axes ], ) -> tuple[ tuple[ float, float ], tuple[ float, float ] ]: + """Gets the limits of both X and Y axis as a 2x2 element tuple. + + Args: + axes (list[axes.Axes]): List of subplots to get limits. + + Returns: + tuple[tuple[float, float], tuple[float, float]]: ((xMin, xMax), (yMin, yMax)) + """ + assert len( axes ) > 0 + xMin, xMax, yMin, yMax = getAxeLimits( axes[ 0 ] ) + if len( axes ) > 1: + for i in range( 1, len( axes ) ): + x1, x2, y1, y2 = getAxeLimits( axes[ i ] ) + if x1 < xMin: + xMin = x1 + if x2 > xMax: + xMax = x2 + if y1 < yMin: + yMin = y1 + if y2 > yMax: + yMax = y2 + return ( ( xMin, xMax ), ( yMin, yMax ) ) + + +def getAxeLimits( ax: axes.Axes ) -> tuple[ float, float, float, float ]: + """Gets the limits of both X and Y axis as a 4 element tuple. + + Args: + ax (axes.Axes): Subplot to get limits. + + Returns: + tuple[float, float, float, float]: (xMin, xMax, yMin, yMax) + """ + xMin, xMax = ax.get_xlim() + yMin, yMax = ax.get_ylim() + return ( xMin, xMax, yMin, yMax ) + + +def findExtremasPropertiesForAssociatedIdentifiers( + df: pd.DataFrame, + associatedIdentifiers: dict[ str, dict[ str, list[ str ] ] ], + offsetPlotting: bool = False, + offsetPercentage: int = 5, +) -> dict[ str, tuple[ float, float ] ]: + """Find min and max of all properties linked to a same identifier. + + Using an associatedIdentifiers dict containing associatedProperties dict, + we can find the extremas for each property of each identifier. Once we have them all, + we compare for each identifier what are the most extreme values and only the biggest and + lowest are kept in the end. + + + Args: + df (pd.DataFrame): Pandas dataframe. + associatedIdentifiers (dict[str, dict[str, list[str]]]): Property identifiers. + offsetPlotting (bool, optional): When using the values being returned, + we might want to add an offset to these values. If set to True, + the offsetPercentage is taken into account. Defaults to False. + offsetPercentage (int, optional): Value by which we will offset + the min and max values of each tuple of floats. Defaults to 5. + + Returns: + dict[str, tuple[float, float]]: { + "BHP (Pa)": (minAllWells, maxAllWells), + "TotalMassRate (kg)": (minAllWells, maxAllWells), + "TotalSurfaceVolumetricRate (m3/s)": (minAllWells, maxAllWells), + "SurfaceVolumetricRateCO2 (m3/s)": (minAllWells, maxAllWells), + "SurfaceVolumetricRateWater (m3/s)": (minAllWells, maxAllWells) + } + """ + extremasProperties: dict[ str, tuple[ float, float ] ] = {} + # First we need to find the extrema for each property type per region. + propertyTypesExtremas: dict[ str, list[ tuple[ float, float ] ] ] = {} + for associatedProperties in associatedIdentifiers.values(): + extremasPerProperty: dict[ str, + tuple[ float, + float ] ] = ( findExtremasAssociatedProperties( df, associatedProperties ) ) + for propertyType, extremaFound in extremasPerProperty.items(): + if propertyType not in propertyTypesExtremas: + propertyTypesExtremas[ propertyType ] = [ extremaFound ] + else: + propertyTypesExtremas[ propertyType ].append( extremaFound ) + # Then, once all extrema have been found for all regions, we need to figure out + # which extrema per property type is the most extreme one. + for propertyType in propertyTypesExtremas: + values: list[ tuple[ float, float ] ] = propertyTypesExtremas[ propertyType ] + minValues: list[ float ] = [ values[ i ][ 0 ] for i in range( len( values ) ) ] + maxValues: list[ float ] = [ values[ i ][ 1 ] for i in range( len( values ) ) ] + lowest, highest = ( min( minValues ), max( maxValues ) ) + if offsetPlotting: + offset: float = ( highest - lowest ) / 100 * offsetPercentage + lowest, highest = ( lowest - offset, highest + offset ) + extremasProperties[ propertyType ] = ( lowest, highest ) + return extremasProperties + + +def findExtremasAssociatedProperties( + df: pd.DataFrame, associatedProperties: dict[ str, list[ str ] ] ) -> dict[ str, tuple[ float, float ] ]: + """Find the min and max of properties. + + Using an associatedProperties dict containing property types + as keys and a list of property names as values, + and a pandas dataframe whose column names are composed of those same + property names, you can find the min and max values of each property + type and return it as a tuple. + + Args: + df (pd.DataFrame): Pandas dataframe. + associatedProperties (dict[str, list[str]]): { + "Pressure (Pa)": ["Reservoir__Pressure__Pa__Source1"], + "Mass (kg)": ["CO2__Mass__kg__Source1", + "Water__Mass__kg__Source1"] + } + + Returns: + dict[str, tuple[float, float]]: { + "Pressure (Pa)": (minPressure, maxPressure), + "Mass (kg)": (minMass, maxMass) + } + """ + extremasProperties: dict[ str, tuple[ float, float ] ] = {} + for propertyType, propertyNames in associatedProperties.items(): + minValues = np.empty( len( propertyNames ) ) + maxValues = np.empty( len( propertyNames ) ) + for i, propertyName in enumerate( propertyNames ): + values: npt.NDArray[ np.float64 ] = df[ propertyName ].to_numpy() + minValues[ i ] = np.nanmin( values ) + maxValues[ i ] = np.nanmax( values ) + extrema: tuple[ float, float ] = ( + float( np.min( minValues ) ), + float( np.max( maxValues ) ), + ) + extremasProperties[ propertyType ] = extrema + return extremasProperties + + +""" +Utils for treatment of the data. +""" + + +def associatePropertyToAxeType( propertyNames: list[ str ] ) -> dict[ str, list[ str ] ]: + """Identify property types. + + From a list of property names, identify if each of this property + corresponds to a certain property type like "Pressure", "Mass", + "Temperature" etc ... and returns a dict where the keys are the property + type and the value the list of property names associated to it. + + Args: + propertyNames (list[str]): ["Reservoir__Pressure__Pa__Source1", + "CO2__Mass__kg__Source1", "Water__Mass__kg__Source1"] + + Returns: + dict[str, list[str]]: { "Pressure (Pa)": ["Reservoir__Pressure__Pa__Source1"], + "Mass (kg)": ["CO2__Mass__kg__Source1", + "Water__Mass__kg__Source1"] } + """ + propertyIds: list[ str ] = fcts.identifyProperties( propertyNames ) + associationTable: dict[ str, str ] = { + "0": "Pressure", + "1": "Pressure", + "2": "Temperature", + "3": "PoreVolume", + "4": "PoreVolume", + "5": "Mass", + "6": "Mass", + "7": "Mass", + "8": "Mass", + "9": "Mass", + "10": "Mass", + "11": "BHP", + "12": "MassRate", + "13": "VolumetricRate", + "14": "VolumetricRate", + "15": "BHP", + "16": "MassRate", + "17": "VolumetricRate", + "18": "VolumetricRate", + "19": "VolumetricRate", + "20": "Volume", + "21": "VolumetricRate", + "22": "Volume", + "23": "Iterations", + "24": "Iterations", + "25": "Stress", + "26": "Displacement", + "27": "Permeability", + "28": "Porosity", + "29": "Ratio", + "30": "Fraction", + "31": "BulkModulus", + "32": "ShearModulus", + "33": "OedometricModulus", + "34": "Points", + "35": "Density", + "36": "Mass", + "37": "Mass", + "38": "Time", + "39": "Time", + } + associatedPropertyToAxeType: dict[ str, list[ str ] ] = {} + noUnitProperties: list[ str ] = [ + "Iterations", + "Porosity", + "Ratio", + "Fraction", + "OedometricModulus", + ] + for i, propId in enumerate( propertyIds ): + idProp: str = propId.split( ":" )[ 0 ] + propNoId: str = propId.split( ":" )[ 1 ] + associatedType: str = associationTable[ idProp ] + if associatedType in noUnitProperties: + axeName: str = associatedType + else: + propIdElts: list[ str ] = propNoId.split( "__" ) + # No unit was found. + if len( propIdElts ) <= 2: + axeName = associatedType + # There is a unit. + else: + unit: str = propIdElts[ -2 ] + axeName = associatedType + " (" + unit + ")" + if axeName not in associatedPropertyToAxeType: + associatedPropertyToAxeType[ axeName ] = [] + associatedPropertyToAxeType[ axeName ].append( propertyNames[ i ] ) + return associatedPropertyToAxeType + + +def propertiesPerIdentifier( propertyNames: list[ str ] ) -> dict[ str, list[ str ] ]: + """Extract identifiers with associated properties. + + From a list of property names, extracts the identifier (name of the + region for flow property or name of a well for well property) and creates + a dictionary with identifiers as keys and the properties containing them + for value in a list. + + Args: + propertyNames (list[str]): Property names. + Example + + .. code-block:: python + + [ + "WellControls1__BHP__Pa__Source1", + "WellControls1__TotalMassRate__kg/s__Source1", + "WellControls2__BHP__Pa__Source1", + "WellControls2__TotalMassRate__kg/s__Source1" + ] + + Returns: + dict[str, list[str]]: Property identifiers. + Example + + .. code-block:: python + + { + "WellControls1": [ + "WellControls1__BHP__Pa__Source1", + "WellControls1__TotalMassRate__kg/s__Source1" + ], + "WellControls2": [ + "WellControls2__BHP__Pa__Source1", + "WellControls2__TotalMassRate__kg/s__Source1" + ] + } + """ + propsPerIdentifier: dict[ str, list[ str ] ] = {} + for propertyName in propertyNames: + elements: list[ str ] = propertyName.split( "__" ) + identifier: str = elements[ 0 ] + if identifier not in propsPerIdentifier: + propsPerIdentifier[ identifier ] = [] + propsPerIdentifier[ identifier ].append( propertyName ) + return propsPerIdentifier + + +def associationIdentifiers( propertyNames: list[ str ] ) -> dict[ str, dict[ str, list[ str ] ] ]: + """Extract identifiers with associated curves. + + From a list of property names, extracts the identifier (name of the + region for flow property or name of a well for well property) and creates + a dictionary with identifiers as keys and the properties containing them + for value in a list. + + Args: + propertyNames (list[str]): Property names. + Example + + .. code-block:: python + + [ + "WellControls1__BHP__Pa__Source1", + "WellControls1__TotalMassRate__kg/s__Source1", + "WellControls1__TotalSurfaceVolumetricRate__m3/s__Source1", + "WellControls1__SurfaceVolumetricRateCO2__m3/s__Source1", + "WellControls1__SurfaceVolumetricRateWater__m3/s__Source1", + "WellControls2__BHP__Pa__Source1", + "WellControls2__TotalMassRate__kg/s__Source1", + "WellControls2__TotalSurfaceVolumetricRate__m3/s__Source1", + "WellControls2__SurfaceVolumetricRateCO2__m3/s__Source1", + "WellControls2__SurfaceVolumetricRateWater__m3/s__Source1", + "WellControls3__BHP__Pa__Source1", + "WellControls3__TotalMassRate__tons/day__Source1", + "WellControls3__TotalSurfaceVolumetricRate__bbl/day__Source1", + "WellControls3__SurfaceVolumetricRateCO2__bbl/day__Source1", + "WellControls3__SurfaceVolumetricRateWater__bbl/day__Source1", + "Mean__BHP__Pa__Source1", + "Mean__TotalMassRate__tons/day__Source1", + "Mean__TotalSurfaceVolumetricRate__bbl/day__Source1", + "Mean__SurfaceVolumetricRateCO2__bbl/day__Source1", + "Mean__SurfaceVolumetricRateWater__bbl/day__Source1" + ] + + Returns: + dict[str, dict[str, list[str]]]: Property identifiers. + Example + + .. code-block:: python + + { + "WellControls1": { + 'BHP (Pa)': [ + 'WellControls1__BHP__Pa__Source1' + ], + 'MassRate (kg/s)': [ + 'WellControls1__TotalMassRate__kg/s__Source1' + ], + 'VolumetricRate (m3/s)': [ + 'WellControls1__TotalSurfaceVolumetricRate__m3/s__Source1', + 'WellControls1__SurfaceVolumetricRateCO2__m3/s__Source1', + 'WellControls1__SurfaceVolumetricRateWater__m3/s__Source1' + ] + }, + "WellControls2": { + 'BHP (Pa)': [ + 'WellControls2__BHP__Pa__Source1' + ], + 'MassRate (kg/s)': [ + 'WellControls2__TotalMassRate__kg/s__Source1' + ], + 'VolumetricRate (m3/s)': [ + 'WellControls2__TotalSurfaceVolumetricRate__m3/s__Source1', + 'WellControls2__SurfaceVolumetricRateCO2__m3/s__Source1', + 'WellControls2__SurfaceVolumetricRateWater__m3/s__Source1' + ] + }, + "WellControls3": { + 'BHP (Pa)': [ + 'WellControls3__BHP__Pa__Source1' + ], + 'MassRate (tons/day)': [ + 'WellControls3__TotalMassRate__tons/day__Source1' + ], + 'VolumetricRate (bbl/day)': [ + 'WellControls3__TotalSurfaceVolumetricRate__bbl/day__Source1', + 'WellControls3__SurfaceVolumetricRateCO2__bbl/day__Source1', + 'WellControls3__SurfaceVolumetricRateWater__bbl/day__Source1' + ] + }, + "Mean": { + 'BHP (Pa)': [ + 'Mean__BHP__Pa__Source1' + ], + 'MassRate (tons/day)': [ + 'Mean__TotalMassRate__tons/day__Source1' + ], + 'VolumetricRate (bbl/day)': [ + 'Mean__TotalSurfaceVolumetricRate__bbl/day__Source1', + 'Mean__SurfaceVolumetricRateCO2__bbl/day__Source1', + 'Mean__SurfaceVolumetricRateWater__bbl/day__Source1' + ] + } + } + """ + propsPerIdentifier: dict[ str, list[ str ] ] = propertiesPerIdentifier( propertyNames ) + assosIdentifier: dict[ str, dict[ str, list[ str ] ] ] = {} + for ident, propNames in propsPerIdentifier.items(): + assosPropsToAxeType: dict[ str, list[ str ] ] = associatePropertyToAxeType( propNames ) + assosIdentifier[ ident ] = assosPropsToAxeType + return assosIdentifier + + +def buildFontTitle( userChoices: dict[ str, Any ] ) -> FontProperties: + """Builds a Fontproperties object according to user choices on title. + + Args: + userChoices (dict[str, Any]): Customization parameters. + + Returns: + FontProperties: FontProperties object for the title. + """ + fontTitle: FontProperties = FontProperties() + if "titleStyle" in userChoices: + fontTitle.set_style( userChoices[ "titleStyle" ] ) + if "titleWeight" in userChoices: + fontTitle.set_weight( userChoices[ "titleWeight" ] ) + if "titleSize" in userChoices: + fontTitle.set_size( userChoices[ "titleSize" ] ) + return fontTitle + + +def buildFontVariable( userChoices: dict[ str, Any ] ) -> FontProperties: + """Builds a Fontproperties object according to user choices on variables. + + Args: + userChoices (dict[str, Any]): Customization parameters. + + Returns: + FontProperties: FontProperties object for the variable axes. + """ + fontVariable: FontProperties = FontProperties() + if "variableStyle" in userChoices: + fontVariable.set_style( userChoices[ "variableStyle" ] ) + if "variableWeight" in userChoices: + fontVariable.set_weight( userChoices[ "variableWeight" ] ) + if "variableSize" in userChoices: + fontVariable.set_size( userChoices[ "variableSize" ] ) + return fontVariable + + +def buildFontCurves( userChoices: dict[ str, Any ] ) -> FontProperties: + """Builds a Fontproperties object according to user choices on curves. + + Args: + userChoices (dict[str, str]): Customization parameters. + + Returns: + FontProperties: FontProperties object for the curves axes. + """ + fontCurves: FontProperties = FontProperties() + if "curvesStyle" in userChoices: + fontCurves.set_style( userChoices[ "curvesStyle" ] ) + if "curvesWeight" in userChoices: + fontCurves.set_weight( userChoices[ "curvesWeight" ] ) + if "curvesSize" in userChoices: + fontCurves.set_size( userChoices[ "curvesSize" ] ) + return fontCurves + + +def customizeLines( userChoices: dict[ str, Any ], labels: list[ str ], + linesList: list[ lines.Line2D ] ) -> list[ lines.Line2D ]: + """Customize lines according to user choices. + + By applying the user choices, we modify or not the list of lines + and return it with the same number of lines in the same order. + + Args: + userChoices (dict[str, Any]): Customization parameters. + labels (list[str]): Labels of lines. + linesList (list[lines.Line2D]): List of lines object. + + Returns: + list[lines.Line2D]: List of lines object modified. + """ + if "linesModified" in userChoices: + linesModified: dict[ str, dict[ str, Any ] ] = userChoices[ "linesModified" ] + linesChanged: list[ lines.Line2D ] = [] + for i, label in enumerate( labels ): + if label in linesModified: + lineChanged: lines.Line2D = applyCustomizationOnLine( linesList[ i ], linesModified[ label ] ) + linesChanged.append( lineChanged ) + else: + linesChanged.append( linesList[ i ] ) + return linesChanged + else: + return linesList + + +def applyCustomizationOnLine( line: lines.Line2D, parameters: dict[ str, Any ] ) -> lines.Line2D: + """Apply modification methods on a line from parameters. + + Args: + line (lines.Line2D): Matplotlib Line2D. + parameters (dict[str, Any]): Dictionary of { + "linestyle": one of ["-","--","-.",":"] + "linewidth": positive int + "color": color code + "marker": one of ["",".","o","^","s","*","D","+","x"] + "markersize":positive int + } + + Returns: + lines.Line2D: Line2D object modified. + """ + if "linestyle" in parameters: + line.set_linestyle( parameters[ "linestyle" ] ) + if "linewidth" in parameters: + line.set_linewidth( parameters[ "linewidth" ] ) + if "color" in parameters: + line.set_color( parameters[ "color" ] ) + if "marker" in parameters: + line.set_marker( parameters[ "marker" ] ) + if "markersize" in parameters: + line.set_markersize( parameters[ "markersize" ] ) + return line + + +""" +Layout tools for layering subplots in a figure. +""" + + +def isprime( x: int ) -> bool: + """Checks if a number is primer or not. + + Args: + x (int): Positive number to test. + + Returns: + bool: True if prime, False if not. + """ + if x < 0: + print( "Invalid number entry, needs to be positive int." ) + return False + + return all( x % n != 0 for n in range( 2, int( x**0.5 ) + 1 ) ) + + +def findClosestPairIntegers( x: int ) -> tuple[ int, int ]: + """Get the pair of integers that multiply the closest to input value. + + Finds the closest pair of integers that when multiplied together, + gives a number the closest to the input number (always above or equal). + + Args: + x (int): Positive number. + + Returns: + tuple[int, int]: (highest int, lowest int) + """ + if x < 4: + return ( x, 1 ) + while isprime( x ): + x += 1 + N: int = round( math.sqrt( x ) ) + while x > N: + if x % N == 0: + M = x // N + highest = max( M, N ) + lowest = min( M, N ) + return ( highest, lowest ) + else: + N += 1 + return ( x, 1 ) + + +def smartLayout( x: int, ratio: float ) -> tuple[ int, int, int ]: + """Return the best layout according to the number of subplots. + + For multiple subplots, we need to have a layout that can adapt to + the number of subplots automatically. This function figures out the + best layout possible knowing the number of suplots and the figure ratio. + + Args: + x (int): Positive number. + ratio (float): Width to height ratio of a figure. + + Returns: + tuple[int]: (nbr_rows, nbr_columns, number of axes to remove) + """ + pair: tuple[ int, int ] = findClosestPairIntegers( x ) + nbrAxesToRemove: int = pair[ 0 ] * pair[ 1 ] - x + if ratio < 1: + return ( pair[ 0 ], pair[ 1 ], nbrAxesToRemove ) + else: + return ( pair[ 1 ], pair[ 0 ], nbrAxesToRemove ) + + +""" +Legend tools +""" + +commonAssociations: dict[ str, str ] = { + "pressuremin": "Pmin", + "pressureMax": "Pmax", + "pressureaverage": "Pavg", + "deltapressuremin": "DPmin", + "deltapressuremax": "DPmax", + "temperaturemin": "Tmin", + "temperaturemax": "Tmax", + "temperatureaverage": "Tavg", + "effectivestressxx": "ESxx", + "effectivestresszz": "ESzz", + "effectivestressratio": "ESratio", + "totaldisplacementx": "TDx", + "totaldisplacementy": "TDy", + "totaldisplacementz": "TDz", + "totalstressXX": "TSxx", + "totalstressZZ": "TSzz", + "stressxx": "Sxx", + "stressyy": "Syy", + "stresszz": "Szz", + "stressxy": "Sxy", + "stressxz": "Sxz", + "stressyz": "Syz", + "poissonratio": "PR", + "porosity": "PORO", + "specificgravity": "SG", + "theoreticalverticalstress": "TVS", + "density": "DNST", + "pressure": "P", + "permeabilityx": "PERMX", + "permeabilityy": "PERMY", + "permeabilityz": "PERMZ", + "oedometric": "OEDO", + "young": "YOUNG", + "shear": "SHEAR", + "bulk": "BULK", + "totaldynamicporevolume": "TDPORV", + "time": "TIME", + "dt": "DT", + "meanbhp": "MBHP", + "meantotalmassrate": "MTMR", + "meantotalvolumetricrate": "MTSVR", + "bhp": "BHP", + "totalmassrate": "TMR", + "cumulatedlineariter": "CLI", + "cumulatednewtoniter": "CNI", + "lineariter": "LI", + "newtoniter": "NI", +} + +phasesAssociations: dict[ str, str ] = { + "dissolvedmass": " IN ", + "immobile": "IMOB ", + "mobile": "MOB ", + "nontrapped": "NTRP ", + "dynamicporevolume": "DPORV ", + "meansurfacevolumetricrate": "MSVR ", + "surfacevolumetricrate": "SVR ", +} + + +def smartLabelsSorted( labels: list[ str ], lines: list[ lines.Line2D ], + userChoices: dict[ str, Any ] ) -> tuple[ list[ str ], list[ lines.Line2D ] ]: + """Shorten all legend labels and sort them. + + To improve readability of the legend for an axe in ParaView, we can apply the + smartLegendLabel functionality to reduce the size of each label. Plus we sort them + alphabetically and therefore, we also sort the lines the same way. + + Args: + labels (list[str]): Labels to use ax.legend() like + ["Region1__TemperatureAvg__K__job_123456", "Region1__PressureMin__Pa__job_123456"] + lines (list[lines.Line2D]): Lines plotted on axes of matplotlib figure like [line1, line2] + userChoices (dict[str, Any]): Choices made by widget selection + in PythonViewConfigurator filter. + + Returns: + tuple[list[str], list[lines.Line2D]]: Improved labels and sorted labels / lines like + (["Region1 Pmin", "Region1 Tavg"], [line2, line1]) + """ + smartLabels: list[ str ] = [ smartLabel( label, userChoices ) for label in labels ] + # I need the labels to be ordered alphabetically for better readability of the legend + # Therefore, if I sort smartLabels, I need to also sort lines with the same order. + # But this can only be done if there are no duplicates of labels in smartLabels. + # If a duplicate is found, "sorted" will try to sort with line which has no comparison built in + # which will throw an error. + if len( set( smartLabels ) ) == len( smartLabels ): + sortedBothLists = sorted( zip( smartLabels, lines, strict=False ) ) + sortedLabels, sortedLines = zip( *sortedBothLists, strict=False ) + return ( list( sortedLabels ), list( sortedLines ) ) + else: + return ( smartLabels, lines ) + + +def smartLabel( label: str, userChoices: dict[ str, Any ] ) -> str: + """Shorten label according to user choices. + + Labels name can tend to be too long. Therefore, we need to reduce the size of the label. + Depending on the choices made by the user, the identifier and the job name can disappear. + + Args: + label (str): A label to be plotted. + Example- Reservoir__DissolvedMassphaseName0InphaseName1__kg__job123456.out + userChoices (dict[str, Any]): user choices. + + Returns: + str: "phaseName0 in phaseName1" or "Reservoir phaseName0 in phaseName1" + or "phaseName0 in phaseName1 job123456.out" or + "Reservoir phaseName0 in phaseName1 job123456.out" + """ + # First step is to abbreviate the label to reduce its size. + smartLabel: str = abbreviateLabel( label ) + # When only one source is used as input, there is no need to precise which one is used + # in the label so the job name is useless. Same when removeJobName option is selected by user. + inputNames: list[ str ] = userChoices[ "inputNames" ] + removeJobName: bool = userChoices[ "removeJobName" ] + if len( inputNames ) > 1 and not removeJobName: + jobName: str = findJobName( label ) + smartLabel += " " + jobName + # When the user chooses to split the plot into subplots to plot by region or well, + # this identifier name will appear as a title of the subplot so no need to use it. + # Same applies when user decides to remove regions. + plotRegions: bool = userChoices[ "plotRegions" ] + removeRegions: bool = userChoices[ "removeRegions" ] + if not plotRegions and not removeRegions: + smartLabel = findIdentifier( label ) + " " + smartLabel + return smartLabel + + +def abbreviateLabel( label: str ) -> str: + """Get the abbreviation of the label according to reservoir nomenclature. + + When using labels to plot, the name can tend to be too long. Therefore, to respect + the logic of reservoir engineering vocabulary, abbreviations for common property names + can be used to shorten the name. The goal is therefore to generate the right abbreviation + for the label input. + + Args: + label (str): A label to be plotted. + Example- Reservoir__DissolvedMassphaseName0InphaseName1__kg__job123456.out + + Returns: + str: "phaseName0 in phaseName1" + """ + for commonAsso in commonAssociations: + if commonAsso in label.lower(): + return commonAssociations[ commonAsso ] + for phaseAsso in phasesAssociations: + if phaseAsso in label.lower(): + phases: list[ str ] = findPhasesLabel( label ) + phase0: str = "" if len( phases ) < 1 else phases[ 0 ] + phase1: str = "" if len( phases ) < 2 else phases[ 1 ] + if phaseAsso == "dissolvedmass": + return phase0 + phasesAssociations[ phaseAsso ] + phase1 + else: + return phasesAssociations[ phaseAsso ] + phase0 + return label + + +def findIdentifier( label: str ) -> str: + """Find identifier inside the label. + + When looking at a label, it may contain or not an identifier at the beginning of it. + An identifier is either a regionName or a wellName. + The goal is to find it and extract it if present. + + Args: + label (str): A label to be plotted. + Example- Reservoir__DissolvedMassphaseName0InphaseName1__kg__job123456.out + + Returns: + str: "Reservoir" + """ + identifier: str = "" + if "__" not in label: + print( "Invalid label, cannot search identifier when no '__' in label." ) + return identifier + subParts: list[ str ] = label.split( "__" ) + if len( subParts ) == 4: + identifier = subParts[ 0 ] + return identifier + + +def findJobName( label: str ) -> str: + """Find the Geos job name at the end of the label. + + When looking at a label, it may contain or not a job name at the end of it. + The goal is to find it and extract it if present. + + Args: + label (str): A label to be plotted. + Example- Reservoir__DissolvedMassphaseName0InphaseName1__kg__job123456.out + + Returns: + str: "job123456.out" + """ + jobName: str = "" + if "__" not in label: + print( "Invalid label, cannot search jobName when no '__' in label." ) + return jobName + subParts: list[ str ] = label.split( "__" ) + if len( subParts ) == 4: + jobName = subParts[ 3 ] + return jobName + + +def findPhasesLabel( label: str ) -> list[ str ]: + """Find phase name inside label. + + When looking at a label, it may contain or not patterns that indicates + the presence of a phase name within it. Therefore, if one of these patterns + is present, one or multiple phase names can be found and be extracted. + + Args: + label (str): A label to be plotted. + Example- Reservoir__DissolvedMassphaseName0InphaseName1__kg__job123456.out + + Returns: + list[str]: [phaseName0, phaseName1] + """ + phases: list[ str ] = [] + lowLabel: str = label.lower() + indexStart: int = 0 + indexEnd: int = 0 + if "__" not in label: + print( "Invalid label, cannot search phases when no '__' in label." ) + return phases + if "dissolvedmass" in lowLabel: + indexStart = lowLabel.index( "dissolvedmass" ) + len( "dissolvedmass" ) + indexEnd = lowLabel.rfind( "__" ) + phasesSubstring: str = lowLabel[ indexStart:indexEnd ] + phases = phasesSubstring.split( "in" ) + phases = [ phase.capitalize() for phase in phases ] + else: + if "dynamicporevolume" in lowLabel: + indexStart = lowLabel.index( "__" ) + 2 + indexEnd = lowLabel.index( "dynamicporevolume" ) + else: + for pattern in [ "nontrapped", "trapped", "immobile", "mobile", "rate" ]: + if pattern in lowLabel: + indexStart = lowLabel.index( pattern ) + len( pattern ) + indexEnd = lowLabel.rfind( "mass" ) + if indexEnd < 0: + indexEnd = indexStart + lowLabel[ indexStart: ].find( "__" ) + break + if indexStart < indexEnd: + phases = [ lowLabel[ indexStart:indexEnd ].capitalize() ] + return phases + + +""" +Under this is the first version of smartLabels without abbreviations. +""" + +# def smartLegendLabelsAndLines( +# labelNames: list[str], lines: list[Any], userChoices: dict[str, Any], regionName="" +# ) -> tuple[list[str], list[Any]]: +# """To improve readability of the legend for an axe in ParaView, we can apply the +# smartLegendLabel functionality to reduce the size of each label. Plus we sort them +# alphabetically and therefore, we also sort the lines the same way. + +# Args: +# labelNames (list[str]): Labels to use ax.legend() like +# ["Region1__PressureMin__Pa__job_123456", "Region1__Temperature__K__job_123456"] +# lines (list[Any]): Lines plotted on axes of matplotlib figure like [line1, line2] +# userChoices (dict[str, Any]): Choices made by widget selection +# in PythonViewConfigurator filter. +# regionName (str, optional): name of the region. Defaults to "". + +# Returns: +# tuple[list[str], list[Any]]: Improved labels and sorted labels / lines like +# (["Temperature K", "PressureMin Pa"], [line2, line1]) +# """ +# smartLabels: list[str] = [ +# smartLegendLabel(labelName, userChoices, regionName) for labelName in labelNames +# ] +# # I need the labels to be ordered alphabetically for better readability of the legend +# # Therefore, if I sort smartLabels, I need to also sort lines with the same order +# sortedBothLists = sorted(zip(smartLabels, lines) +# sortedLabels, sortedLines = zip(*sortedBothLists) +# return (sortedLabels, sortedLines) + +# def smartLegendLabel(labelName: str, userChoices: dict[str, Any], regionName="") -> str: +# """When plotting legend label, the label format can be improved by removing some +# overwhelming / repetitive prefix / suffix and have a shorter label. + +# Args: +# labelName (str): Label to use ax.legend() like +# Region1__PressureMin__Pa__job_123456 +# userChoices (dict[str, Any]): Choices made by widget selection +# in PythonViewConfigurator filter. +# regionName (str, optional): name of the region. Defaults to "". + +# Returns: +# str: Improved label name like PressureMin Pa. +# """ +# smartLabel: str = "" +# # When only one source is used as input, there is no need to precise which one +# # is used in the label. Same when removeJobName option is selected by user. +# inputNames: list[str] = userChoices["inputNames"] +# removeJobName: bool = userChoices["removeJobName"] +# if len(inputNames) <= 1 or removeJobName: +# smartLabel = removeJobNameInLegendLabel(labelName, inputNames) +# # When the user chooses to split the plot into subplots to plot by region, +# # the region name will appear as a title of the subplot so no need to use it. +# # Same applies when user decides to remove regions. +# plotRegions: bool = userChoices["plotRegions"] +# removeRegions: bool = userChoices["removeRegions"] +# if plotRegions or removeRegions: +# smartLabel = removeIdentifierInLegendLabel(smartLabel, regionName) +# smartLabel = smartLabel.replace("__", " ") +# return smartLabel + +# def removeJobNameInLegendLabel(legendLabel: str, inputNames: list[str]) -> str: +# """When plotting legends, the name of the job is by default at the end of +# the label. Therefore, it can increase tremendously the size of the legend +# and we can avoid that by removing the job name from it. + +# Args: +# legendLabel (str): Label to use ax.legend() like +# Region1__PressureMin__Pa__job_123456 +# inputNames (list[str]): names of the sources use to plot. + +# Returns: +# str: Label without the job name like Region1__PressureMin__Pa. +# """ +# for inputName in inputNames: +# pattern: str = "__" + inputName +# if legendLabel.endswith(pattern): +# jobIndex: int = legendLabel.index(pattern) +# return legendLabel[:jobIndex] +# return legendLabel + +# def removeIdentifierInLegendLabel(legendLabel: str, regionName="") -> str: +# """When plotting legends, the name of the region is by default at the +# beginning of the label. Here we remove the region name from the legend label. + +# Args: +# legendLabel (str): Label to use ax.legend() like +# Region1__PressureMin__Pa__job_123456 +# regionName (str): name of the region. Defaults to "". + +# Returns: +# str: Label without the job name like PressureMin__Pa__job_123456 +# """ +# if "__" not in legendLabel: +# return legendLabel +# if regionName == "": +# firstRegionIndex: int = legendLabel.index("__") +# return legendLabel[firstRegionIndex + 2:] +# pattern: str = regionName + "__" +# if legendLabel.startswith(pattern): +# return legendLabel[len(pattern):] +# return legendLabel + +""" +Other 2D tools for simplest figures +""" + + +def basicFigure( df: pd.DataFrame, variableName: str, curveName: str ) -> tuple[ figure.Figure, axes.Axes ]: + """Creates a plot. + + Generates a figure and axes objects from matplotlib that plots + one curve along the X axis, with legend and label for X and Y. + + Args: + df (pd.DataFrame): Dataframe containing at least two columns, + one named "variableName" and the other "curveName". + variableName (str): Name of the variable column. + curveName (str): Name of the column to display along that variable. + + Returns: + tuple[figure.Figure, axes.Axes]: The fig and the ax. + """ + fig, ax = plt.subplots() + x: npt.NDArray[ np.float64 ] = df[ variableName ].to_numpy() + y: npt.NDArray[ np.float64 ] = df[ curveName ].to_numpy() + ax.plot( x, y, label=curveName ) + ax.set_xlabel( variableName ) + ax.set_ylabel( curveName ) + ax.legend( loc="best" ) + return ( fig, ax ) + + +def invertedBasicFigure( df: pd.DataFrame, variableName: str, curveName: str ) -> tuple[ figure.Figure, axes.Axes ]: + """Creates a plot with inverted XY axis. + + Generates a figure and axes objects from matplotlib that plots + one curve along the Y axis, with legend and label for X and Y. + + Args: + df (pd.DataFrame): Dataframe containing at least two columns, + one named "variableName" and the other "curveName". + variableName (str): Name of the variable column. + curveName (str): Name of the column to display along that variable. + + Returns: + tuple[figure.Figure, axes.Axes]: The fig and the ax. + """ + fig, ax = plt.subplots() + x: npt.NDArray[ np.float64 ] = df[ curveName ].to_numpy() + y: npt.NDArray[ np.float64 ] = df[ variableName ].to_numpy() + ax.plot( x, y, label=variableName ) + ax.set_xlabel( curveName ) + ax.set_ylabel( variableName ) + ax.legend( loc="best" ) + return ( fig, ax ) + + +def adjust_subplots( fig: figure.Figure, invertXY: bool ) -> figure.Figure: + """Adjust the size of the subplot in the fig. + + Args: + fig (figure.Figure): Matplotlib figure. + invertXY (bool): Choice to either swap or not the X and Y axes. + + Returns: + figure.Figure: Matplotlib figure with adjustments. + """ + if invertXY: + fig.subplots_adjust( left=0.05, right=0.98, top=0.9, bottom=0.2 ) + else: + fig.subplots_adjust( left=0.06, right=0.94, top=0.95, bottom=0.08 ) + return fig diff --git a/geos-pv/src/geos/pv/pythonViewUtils/mainPythonView.py b/geos-pv/src/geos/pv/pythonViewUtils/mainPythonView.py new file mode 100755 index 00000000..bc7e7818 --- /dev/null +++ b/geos-pv/src/geos/pv/pythonViewUtils/mainPythonView.py @@ -0,0 +1,45 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright 2023-2024 TotalEnergies. +# SPDX-FileContributor: Alexandre Benedicto +# type: ignore +# ruff: noqa +from logging import Logger, getLogger, INFO +from paraview.detail.loghandler import ( # type: ignore[import-not-found] + VTKHandler, +) # source: https://github.com/Kitware/ParaView/blob/master/Wrapping/Python/paraview/detail/loghandler.py + +logger: Logger = getLogger( "Python View Configurator" ) +logger.setLevel( INFO ) +vtkHandler: VTKHandler = VTKHandler() +logger.addHandler( vtkHandler ) + +try: + import matplotlib.pyplot as plt + from paraview import python_view + + import geos.pv.utils.paraviewTreatments as pvt + from geos.pv.pythonViewUtils.Figure2DGenerator import ( + Figure2DGenerator, ) + + plt.close() + if len( sourceNames ) == 0: # noqa: F821 + raise ValueError( "No source name was found. Please check at least one source in <>." ) + + dataframes = pvt.getDataframesFromMultipleVTKSources( + sourceNames, + variableName # noqa: F821 + ) + dataframe = pvt.mergeDataframes( dataframes, variableName ) # noqa: F821 + obj_figure = Figure2DGenerator( dataframe, userChoices, logger ) # noqa: F821 + fig = obj_figure.getFigure() + + def setup_data( view ) -> None: # noqa + pass + + def render( view, width: int, height: int ): # noqa + fig.set_size_inches( float( width ) / 100.0, float( height ) / 100.0 ) + imageToReturn = python_view.figure_to_image( fig ) + return imageToReturn + +except Exception as e: + logger.critical( e, exc_info=True )