feat: Derive all PyFluent warnings from common base classes. (#2852)

* feat: Derive all PyFluent warnings from common base classes.

* feat: Derive all PyFluent warnings from common base classes.

* fix: test

src/ansys/fluent/core/__init__.py

@@ -33,7 +33,11 @@
 from ansys.fluent.core.utils.fluent_version import FluentVersion  # noqa: F401
 from ansys.fluent.core.utils.search import search  # noqa: F401
 from ansys.fluent.core.utils.setup_for_fluent import setup_for_fluent  # noqa: F401
-from ansys.fluent.core.warnings import PyFluentDeprecationWarning  # noqa: F401
+from ansys.fluent.core.warnings import (  # noqa: F401
+    PyFluentDeprecationWarning,
+    PyFluentUserWarning,
+    warning,
+)
 
 _VERSION_INFO = None
 """Global variable indicating the version of the PyFluent package - Empty by default"""

src/ansys/fluent/core/examples/downloads.py

@@ -12,6 +12,7 @@
 import requests
 
 import ansys.fluent.core as pyfluent
+from ansys.fluent.core.warnings import PyFluentUserWarning
 
 logger = logging.getLogger("pyfluent.networking")
 
@@ -71,7 +72,8 @@ def _retrieve_file(
     logger.info(f"Checking if {local_path_no_zip} already exists...")
     if os.path.isfile(local_path_no_zip) or os.path.isdir(local_path_no_zip):
         warnings.warn(
-            f"\nFile already exists. File path:\n{local_path_no_zip}\n", UserWarning
+            f"\nFile already exists. File path:\n{local_path_no_zip}\n",
+            PyFluentUserWarning,
         )
         logger.info("File already exists.")
         if return_without_path:

src/ansys/fluent/core/fluent_connection.py

@@ -21,6 +21,7 @@
 from ansys.fluent.core.services.scheme_eval import SchemeEvalService
 from ansys.fluent.core.utils.execution import timeout_exec, timeout_loop
 from ansys.fluent.core.utils.file_transfer_service import RemoteFileTransferStrategy
+from ansys.fluent.core.warnings import PyFluentDeprecationWarning
 from ansys.platform.instancemanagement import Instance
 import docker
 
@@ -546,7 +547,7 @@ def create_grpc_service(self, service, *args):
 
     def check_health(self) -> str:
         """Check health of Fluent connection."""
-        warnings.warn("Use -> health_check.status()", DeprecationWarning)
+        warnings.warn("Use -> health_check.status()", PyFluentDeprecationWarning)
         return self.health_check.status()
 
     def wait_process_finished(self, wait: Union[float, int, bool] = 60):

src/ansys/fluent/core/services/solution_variables.py

@@ -19,6 +19,7 @@
     TracingInterceptor,
 )
 from ansys.fluent.core.solver.error_message import allowed_name_error_message
+from ansys.fluent.core.warnings import PyFluentDeprecationWarning
 
 
 class SolutionVariableService:
@@ -125,7 +126,7 @@ def svars(self) -> List[str]:
             """Solution variables."""
             warnings.warn(
                 "svars is deprecated, use solution_variables instead",
-                DeprecationWarning,
+                PyFluentDeprecationWarning,
             )
             return self.solution_variables
 
@@ -240,7 +241,7 @@ def get_svars_info(
         """Get solution variables info."""
         warnings.warn(
             "get_svars_info is deprecated, use get_variables_info instead",
-            DeprecationWarning,
+            PyFluentDeprecationWarning,
         )
         return self.get_variables_info(zone_names=zone_names, domain_name=domain_name)
 
@@ -616,7 +617,7 @@ def get_svar_data(
         """Get solution variable data."""
         warnings.warn(
             "get_svar_data is deprecated, use get_data instead",
-            DeprecationWarning,
+            PyFluentDeprecationWarning,
         )
         return self.get_data(
             solution_variable_name=svar_name,
@@ -733,7 +734,7 @@ def set_svar_data(
         """Set solution variable data."""
         warnings.warn(
             "set_svar_data is deprecated, use set_data instead",
-            DeprecationWarning,
+            PyFluentDeprecationWarning,
         )
         return self.set_data(
             solution_variable_name=svar_name,

src/ansys/fluent/core/session.py

@@ -23,6 +23,7 @@
 from ansys.fluent.core.streaming_services.transcript_streaming import Transcript
 from ansys.fluent.core.utils import load_module
 from ansys.fluent.core.utils.fluent_version import FluentVersion
+from ansys.fluent.core.warnings import PyFluentDeprecationWarning, PyFluentUserWarning
 
 from .rpvars import RPVars
 
@@ -234,7 +235,7 @@ def field_info(self):
         """Provides access to Fluent field information."""
         warnings.warn(
             "field_info is deprecated. Use fields.field_info instead.",
-            DeprecationWarning,
+            PyFluentDeprecationWarning,
         )
         return self.fields.field_info
 
@@ -243,7 +244,7 @@ def field_data(self):
         """Fluent field data on surfaces."""
         warnings.warn(
             "field_data is deprecated. Use fields.field_data instead.",
-            DeprecationWarning,
+            PyFluentDeprecationWarning,
         )
         return self.fields.field_data
 
@@ -252,7 +253,7 @@ def field_data_streaming(self):
         """Field gRPC streaming service of Fluent."""
         warnings.warn(
             "field_data_streaming is deprecated. Use fields.field_data_streaming instead.",
-            DeprecationWarning,
+            PyFluentDeprecationWarning,
         )
         return self.fields.field_data_streaming
 
@@ -263,12 +264,12 @@ def id(self) -> str:
 
     def start_journal(self, file_name: str):
         """Executes tui command to start journal."""
-        warnings.warn("Use -> journal.start()", DeprecationWarning)
+        warnings.warn("Use -> journal.start()", PyFluentDeprecationWarning)
         self.journal.start(file_name)
 
     def stop_journal(self):
         """Executes tui command to stop journal."""
-        warnings.warn("Use -> journal.stop()", DeprecationWarning)
+        warnings.warn("Use -> journal.stop()", PyFluentDeprecationWarning)
         self.journal.stop()
 
     @classmethod
@@ -375,7 +376,7 @@ def upload(
         remote_file_name : str, optional
             remote file name, by default None
         """
-        warnings.warn(self._file_transfer_api_warning("upload()"), UserWarning)
+        warnings.warn(self._file_transfer_api_warning("upload()"), PyFluentUserWarning)
         if self._file_transfer_service:
             return self._file_transfer_service.upload(file_name, remote_file_name)
 
@@ -389,7 +390,9 @@ def download(self, file_name: str, local_directory: Optional[str] = "."):
         local_directory : str, optional
             Local destination directory. The default is the current working directory.
         """
-        warnings.warn(self._file_transfer_api_warning("download()"), UserWarning)
+        warnings.warn(
+            self._file_transfer_api_warning("download()"), PyFluentUserWarning
+        )
         if self._file_transfer_service:
             return self._file_transfer_service.download(file_name, local_directory)
 

src/ansys/fluent/core/session_solver.py

@@ -34,6 +34,7 @@
     FluentVersion,
     get_version_for_file_name,
 )
+from ansys.fluent.core.warnings import PyFluentDeprecationWarning
 from ansys.fluent.core.workflow import ClassicWorkflow
 
 tui_logger = logging.getLogger("pyfluent.tui")
@@ -147,7 +148,7 @@ def svar_data(self):
         """``SolutionVariableData`` handle."""
         warnings.warn(
             "svar_data is deprecated. Use fields.solution_variable_data instead.",
-            DeprecationWarning,
+            PyFluentDeprecationWarning,
         )
         return self.fields.solution_variable_data
 
@@ -156,7 +157,7 @@ def svar_info(self):
         """``SolutionVariableInfo`` handle."""
         warnings.warn(
             "svar_info is deprecated. Use fields.solution_variable_info instead.",
-            DeprecationWarning,
+            PyFluentDeprecationWarning,
         )
         return self.fields.solution_variable_info
 
@@ -165,7 +166,7 @@ def reduction(self):
         """``Reduction`` handle."""
         warnings.warn(
             "reduction is deprecated. Use fields.reduction instead.",
-            DeprecationWarning,
+            PyFluentDeprecationWarning,
         )
         return self.fields.reduction
 

src/ansys/fluent/core/solver/flobject.py

@@ -49,12 +49,18 @@
 from zipimport import zipimporter
 
 try:
+    from ansys.fluent.core.warnings import (
+        PyFluentDeprecationWarning,
+        PyFluentUserWarning,
+    )
     import ansys.units as ansys_units
 
     from .flunits import UnhandledQuantity, get_si_unit_for_fluent_quantity
 except ImportError:
     get_unit_for_fl_quantity_attr = None
     ansys_units = None
+    PyFluentDeprecationWarning = FutureWarning
+    PyFluentUserWarning = UserWarning
 
 from .error_message import allowed_name_error_message, allowed_values_error
 
@@ -519,13 +525,13 @@ class Textual(Property):
     """Exposes attribute accessor on settings object - specific to string objects."""
 
 
-class DeprecatedSettingWarning(FutureWarning):
+class DeprecatedSettingWarning(PyFluentDeprecationWarning):
     """Provides deprecated settings warning."""
 
     pass
 
 
-class UnstableSettingWarning(UserWarning):
+class UnstableSettingWarning(PyFluentUserWarning):
     """Provides unstable settings warning."""
 
     pass

src/ansys/fluent/core/utils/deprecate_args.py

@@ -5,12 +5,13 @@
 import warnings
 
 from ansys.fluent.core.launcher.pyfluent_enums import FluentEnum
+from ansys.fluent.core.warnings import PyFluentDeprecationWarning
 
 logger = logging.getLogger("pyfluent.general")
 
 
 def deprecate_argument(
-    old_arg, new_arg, converter, deprecation_class=DeprecationWarning
+    old_arg, new_arg, converter, deprecation_class=PyFluentDeprecationWarning
 ):
     """Warns that the argument provided is deprecated and automatically replaces the
     deprecated argument with the appropriate new argument."""

src/ansys/fluent/core/utils/file_transfer_service.py

@@ -11,6 +11,7 @@
 from alive_progress import alive_bar
 
 import ansys.fluent.core as pyfluent
+from ansys.fluent.core.warnings import PyFluentUserWarning
 import ansys.platform.instancemanagement as pypim
 import ansys.tools.filetransfer as ft
 import docker
@@ -218,7 +219,7 @@ def upload(
                 if is_file_on_remote:
                     warnings.warn(
                         f"\n{file} with the same name exists at the remote location.\n",
-                        UserWarning,
+                        PyFluentUserWarning,
                     )
                 elif os.path.isfile(file) and not is_file_on_remote:
                     self.client.upload_file(
@@ -249,7 +250,8 @@ def download(
             for file in files:
                 if os.path.isfile(file):
                     warnings.warn(
-                        f"\nFile already exists. File path:\n{file}\n", UserWarning
+                        f"\nFile already exists. File path:\n{file}\n",
+                        PyFluentUserWarning,
                     )
                 else:
                     self.client.download_file(
@@ -388,7 +390,7 @@ def upload(
                         else:
                             warnings.warn(
                                 f"\n{file} with the same name exists at the remote location.\n",
-                                UserWarning,
+                                PyFluentUserWarning,
                             )
                     elif not self.file_service.file_exist(os.path.basename(file)):
                         raise FileNotFoundError(f"{file} does not exist.")
@@ -436,7 +438,8 @@ def download(
                 for file in files:
                     if os.path.isfile(file):
                         warnings.warn(
-                            f"\nFile already exists. File path:\n{file}\n", UserWarning
+                            f"\nFile already exists. File path:\n{file}\n",
+                            PyFluentUserWarning,
                         )
                     else:
                         self.download_file(

src/ansys/fluent/core/warnings.py

@@ -1,7 +1,33 @@
 """Provides a module to get warnings for core functionality."""
 
+import warnings
+
 
 class PyFluentDeprecationWarning(FutureWarning):
-    """Provides the common warning class for all deprecated PyFluent feature."""
+    """Provides the common warning class for warnings about deprecated PyFluent
+    features."""
+
+    pass
+
+
+class PyFluentUserWarning(UserWarning):
+    """Provides the common warning class for warnings generated from user code."""
 
     pass
+
+
+class WarningControl:
+    """Class to control warnings in PyFluent."""
+
+    def enable(self):
+        """Enables all PyFluent warnings."""
+        warnings.simplefilter("default", PyFluentDeprecationWarning)
+        warnings.simplefilter("default", PyFluentUserWarning)
+
+    def disable(self):
+        """Disables all PyFluent warnings."""
+        warnings.simplefilter("ignore", PyFluentDeprecationWarning)
+        warnings.simplefilter("ignore", PyFluentUserWarning)
+
+
+warning = WarningControl()

src/ansys/fluent/core/workflow.py

@@ -16,6 +16,7 @@
 )
 from ansys.fluent.core.utils.dictionary_operations import get_first_dict_key_for_value
 from ansys.fluent.core.utils.fluent_version import FluentVersion
+from ansys.fluent.core.warnings import PyFluentDeprecationWarning, PyFluentUserWarning
 
 
 class CommandInstanceCreationError(RuntimeError):
@@ -952,7 +953,7 @@ def command_arguments(self) -> ReadOnlyObject:
         ReadOnlyObject
             The task's arguments.
         """
-        warnings.warn("CommandArguments", DeprecationWarning)
+        warnings.warn("CommandArguments", PyFluentDeprecationWarning)
         return self._refreshed_command()
 
     @property
@@ -1080,7 +1081,7 @@ def command_arguments(self) -> ReadOnlyObject:
         ReadOnlyObject
             The task's arguments.
         """
-        warnings.warn("CommandArguments", DeprecationWarning)
+        warnings.warn("CommandArguments", PyFluentDeprecationWarning)
         return {}
 
     @property
@@ -1193,7 +1194,7 @@ def add_child_and_update(self, state=None, defer_update=None):
             if defer_update is not None:
                 warnings.warn(
                     " The 'defer_update()' method is supported in Fluent 2024 R1 and later.",
-                    UserWarning,
+                    PyFluentUserWarning,
                 )
             self._task.AddChildAndUpdate()
         return self.last_child()

tests/test_session.py

@@ -26,6 +26,7 @@
 from ansys.fluent.core.utils.file_transfer_service import RemoteFileTransferStrategy
 from ansys.fluent.core.utils.fluent_version import FluentVersion
 from ansys.fluent.core.utils.networking import get_free_port
+from ansys.fluent.core.warnings import PyFluentDeprecationWarning
 
 
 class MockSettingsServicer(settings_pb2_grpc.SettingsServicer):
@@ -487,9 +488,9 @@ def test_get_set_state_on_solver(new_solver_session):
 
 def test_solver_structure(new_solver_session):
     solver = new_solver_session
-    with pytest.warns(DeprecationWarning):
+    with pytest.warns(PyFluentDeprecationWarning):
         solver.field_data
-    with pytest.warns(DeprecationWarning):
+    with pytest.warns(PyFluentDeprecationWarning):
         solver.svar_data
 
     assert {