Feat/Richer cmd output for listing commands

ansys/mapdl/core/commands.py

@@ -20,6 +20,160 @@
     inq_func
 )
 
+import re
+import numpy as np
+
+MSG_NOT_PANDAS = """'Pandas' is not installed or could not be found.
+Hence this command is not applicable.
+
+You can install it using:
+pip install pandas
+"""
+
+# Identify where the data start in the output
+GROUP_DATA_START = ['NODE', 'ELEM']
+
+# Allowed commands to get output as array or dataframe.
+# In theory, these commands should follow the same format.
+# Some of them are not documented (already deprecated?)
+# So they won't be wrapped.
+CMD_LISTING = [
+    'NLIN', # not documented
+    'PRCI',
+    'PRDI', # Not documented.
+    'PREF', # Not documented.
+    'PREN',
+    'PRER',
+    'PRES',
+    'PRET',
+    'PRGS', # Not documented.
+    'PRIN',
+    'PRIT',
+    'PRJS',
+    'PRNL',
+    'PRNM', # Not documented.
+    'PRNS',
+    'PROR',
+    'PRPA',
+    'PRRF',
+    'PRRS',
+    'PRSE',
+    'PRSS', # Not documented.
+    'PRST', # Not documented.
+    'PRVE',
+    'PRXF', # Not documented.
+    'STAT',
+    'SWLI'
+]
+
+# Adding empty lines to match current format.
+docstring_injection = """
+Returns
+-------
+
+str
+    Str object with the command console output.
+
+    This object also has the extra methods:
+
+    * ``str.to_list()``
+
+    * ``str.to_array()`` (Only on listing commands)
+
+    * ``str.to_dataframe()`` (Only if Pandas is installed)
+
+    For more information visit `PyMAPDL Post-Processing <https://mapdldocs.pyansys.com/user_guide/post.html>`_.
+
+"""
+
+
+def get_indentation(indentation_regx, docstring):
+    return re.findall(indentation_regx, docstring, flags=re.DOTALL|re.IGNORECASE)[0][0]
+
+
+def indent_text(indentation, docstring_injection):
+    return '\n'.join([indentation + each for each in docstring_injection.splitlines() if each.strip()])
+    # return '\n'.join([indentation + each if each.strip() else '' for each in docstring_injection.splitlines()])
+
+
+def get_docstring_indentation(docstring):
+    indentation_regx = r'\n(\s*)\n'
+    return get_indentation(indentation_regx, docstring)
+
+
+def get_sections(docstring):
+    return [each.strip().lower() for each in re.findall(r'\n\s*(\S*)\n\s*-+\n', docstring)]
+
+
+def get_section_indentation(section_name, docstring):
+    sections = get_sections(docstring)
+    if section_name.lower().strip() not in sections:
+        raise ValueError(f"This section '{section_name.lower().strip()}' does not exist in the docstring.")
+    section_match = section_name + r'\n\s*-*'
+    indent_match = r'\n(\s*)(\S)'
+    indentation_regx = section_match + indent_match
+    return get_indentation(indentation_regx, docstring)
+
+
+def inject_before(section, indentation, indented_doc_inject, docstring):
+    return re.sub(section + r'\n\s*-*', f"{indented_doc_inject.strip()}\n\n{indentation}\g<0>", docstring, flags=re.IGNORECASE)
+
+
+def inject_after_return_section(indented_doc_inject, docstring):
+    return re.sub('Returns' + r'\n\s*-*', f"{indented_doc_inject.strip()}\n", docstring, flags=re.IGNORECASE)
+
+
+def inject_docs(docstring):
+    """Inject a string in a docstring"""
+    return_header = r'Returns\n\s*-*'
+    if re.search(return_header, docstring):
+        # There is a return block already, probably it should not.
+        indentation = get_section_indentation('Returns', docstring)
+        indented_doc_inject = indent_text(indentation, docstring_injection)
+        return inject_after_return_section(indented_doc_inject, docstring)
+    else:
+        # There is not returns header
+        # find sections
+        sections = get_sections(docstring)
+
+        if 'parameters' in sections:
+            ind = sections.index('parameters')
+            if ind == len(sections) - 1:
+                # The parameters is the last bit. Just append it.
+                indentation = get_section_indentation('Parameters', docstring)
+                indented_doc_inject = indent_text(indentation, docstring_injection)
+                return docstring + '\n' + indented_doc_inject
+            else:
+                # inject it right before the section after 'parameter'
+                sect_after_parameter = sections[ind + 1]
+                indentation = get_section_indentation(sect_after_parameter, docstring)
+                indented_doc_inject = indent_text(indentation, docstring_injection)
+                return inject_before(sect_after_parameter, indentation, indented_doc_inject, docstring)
+
+        elif 'notes' in sections:
+            indentation = get_section_indentation('Notes', docstring)
+            indented_doc_inject = indent_text(indentation, docstring_injection)
+            return inject_before('Notes', indentation, indented_doc_inject, docstring)
+
+        else:
+            indentation = get_docstring_indentation(docstring)
+            indented_doc_inject = indent_text(indentation, docstring_injection)
+            return docstring + '\n' + indented_doc_inject
+
+
+def check_valid_output(func):
+    """Wrapper that check if output can be wrapped by pandas, if not, it will raise an exception."""
+
+    def func_wrapper(self, *args, **kwargs):
+        output = self.__str__()
+        if '*** WARNING ***' in output or '*** ERROR ***' in output: # Error should be caught in mapdl.run.
+            err_type = re.findall('\*\*\* (.*) \*\*\*', output)[0]
+            msg = f'Unable to parse because of next {err_type.title()}' + '\n'.join(output.splitlines()[-2:])
+            raise ValueError(msg)
+        else:
+            return func(self, *args, **kwargs)
+    return func_wrapper
+
 
 class PreprocessorCommands(
     preproc.database.Database,
@@ -219,7 +373,13 @@ class Commands(
 
 
 class CommandOutput(str):
-    """Customized command output."""
+    """Custom string subclass for handling the commands output.
+
+    This class add two method to track the cmd which generated this output.
+    * ``cmd`` - The MAPDL command which generated the output.
+    * ``command`` - The full command line (with arguments) which generated the output.
+
+    """
 
     ## References:
     # - https://stackoverflow.com/questions/7255655/how-to-subclass-str-in-python
@@ -249,3 +409,162 @@ def command(self):
     def command(self):
         """Not allowed to change the value of ``command``."""
         pass
+
+
+class CommandListingOutput(CommandOutput):
+    """Allow the conversion of command output to native Python types.
+
+    Custom class for handling the commands whose output is sensible to be converted to
+    a list of lists, a Numpy array or a Pandas DataFrame.
+    """
+
+    def _is_data_start(self, line, magicword=None):
+        """Check if line is the start of a data group."""
+        if not magicword:
+            magicword = GROUP_DATA_START
+
+        # Checking if we are supplying a custom start function.
+        if self.custom_data_start(line) is not None:
+            return self.custom_data_start(line)
+
+        if line.split():
+            if line.split()[0] in magicword or self.custom_data_start(line):
+                return True
+        return False
+
+    def _is_data_end(self, line):
+        """Check if line is the end of a data group."""
+
+        # Checking if we are supplying a custom start function.
+        if self.custom_data_end(line) is not None:
+            return self.custom_data_end(line)
+        else:
+            return self._is_empty(line)
+
+    def custom_data_start(self, line):
+        """Custom data start line check function.
+
+        This function is left empty so it can be overwritten by the user.
+
+        If modified, it should return ``True`` when the line is the start
+        of a data group, otherwise it should return ``False``.
+        """
+        return None
+
+    def custom_data_end(self, line):
+        """Custom data end line check function.
+
+        This function is left empty so it can be overwritten by the user.
+
+        If modified, it should return ``True`` when the line is the end
+        of a data group, otherwise it should return ``False``.
+        """
+        return None
+
+    @staticmethod
+    def _is_empty_line(line):
+        return bool(line.split())
+
+    def _format(self):
+        """Perform some formatting (replacing mainly) in the raw text."""
+        return re.sub(r'[^E](-)', ' -', self.__str__())
+
+    def _get_body(self, trail_header=None):
+        """Get command body text.
+
+        It removes the maximum absolute values tail part and makes sure there is separation between columns.
+        """
+        # Doing some formatting of the string
+        body = self._format().splitlines()
+
+        if not trail_header:
+            trail_header = ['MAXIMUM ABSOLUTE VALUES', 'TOTAL VALUES']
+
+        if not isinstance(trail_header, list):
+            trail_header = list(trail_header)
+
+        # Removing parts matching trail_header
+        for each_trail_header in trail_header:
+            if each_trail_header in self.__str__():
+                # starting to check from the bottom.
+                for i in range(len(body) - 1, -1, -1):
+                    if each_trail_header in body[i]:
+                        break
+                body = body[:i]
+        return body
+
+    def _get_data_group_indexes(self, body, magicword=None):
+        """Return the indexes of the start and end of the data groups."""
+
+        if '*****ANSYS VERIFICATION RUN ONLY*****' in str(self):
+            shift = 2
+        else:
+            shift = 0
+
+        # Getting pairs of starting end
+        start_idxs = [ind for ind, each in enumerate(body) if self._is_data_start(each, magicword=magicword)]
+        end_idxs = [ind - shift for ind, each in enumerate(body) if self._is_empty_line(each)]
+
+        indexes = [*start_idxs, *end_idxs]
+        indexes.sort()
+
+        ends = [indexes[indexes.index(each) + 1] for each in start_idxs[:-1]]
+        ends.append(len(body))
+
+        return zip(start_idxs, ends)
+
+    def _get_data_groups(self, magicword=None, trail_header=None):
+        """Get raw data groups"""
+        body = self._get_body(trail_header=trail_header)
+
+        data = []
+        for start, end in self._get_data_group_indexes(body, magicword=magicword):
+            data.extend(body[start+1:end])
+
+        # removing empty lines
+        return [each for each in data if each]
+
+    def get_columns(self):
+        """Get the column names for the dataframe.
+
+        Returns
+        -------
+        List of strings
+
+        """
+        body = self._get_body()
+        pairs = list(self._get_data_group_indexes(body))
+        return body[pairs[0][0]].split()
+
+    @check_valid_output
+    def to_list(self):
+        """Export the command output a list or list of lists.
+
+        Returns
+        -------
+            List of strings
+        """
+        data = self._get_data_groups()
+        return [each.split() for each in data]
+
+    def to_array(self):
+        """Export the command output as a numpy array.
+
+        Returns
+        -------
+            Numpy array
+        """
+        return np.array(self.to_list(), dtype=float)
+
+    def to_dataframe(self):
+        """Export the command output as a Pandas DataFrame.
+
+        Returns
+        -------
+            Pandas Dataframe
+        """
+        try:
+            import pandas as pd
+        except ModuleNotFoundError:
+            raise ModuleNotFoundError(MSG_NOT_PANDAS)
+        return pd.DataFrame(data=self.to_array(), columns=self.get_columns())

ansys/mapdl/core/mapdl.py

@@ -25,7 +25,7 @@
 from ansys.mapdl.core.errors import MapdlRuntimeError, MapdlInvalidRoutineError
 from ansys.mapdl.core.plotting import general_plotter
 from ansys.mapdl.core.post import PostProcessing
-from ansys.mapdl.core.commands import Commands
+from ansys.mapdl.core.commands import Commands, CommandListingOutput, CMD_LISTING, inject_docs
 from ansys.mapdl.core.inline_functions import Query
 from ansys.mapdl.core import LOG as logger
 from ansys.mapdl.reader.rst import Result
@@ -165,6 +165,23 @@ def __init__(self, loglevel='DEBUG', use_vtk=True, log_apdl=None,
 
         self._post = PostProcessing(self)
 
+        self._wrap_listing_functions()
+
+    def _wrap_listing_functions(self):
+        # Wrapping LISTING FUNCTIONS.
+        def wrap_listing_function(func):
+            # Injecting doc string modification
+            func.__func__.__doc__ = inject_docs(func.__func__.__doc__)
+            @wraps(func)
+            def inner_wrapper(*args, **kwargs):
+                return CommandListingOutput(func(*args, **kwargs))
+            return inner_wrapper
+
+        for name in dir(self):
+            if name[0:4].upper() in CMD_LISTING:
+                func = self.__getattribute__(name)
+                setattr(self, name, wrap_listing_function(func))
+
     @property
     def _name(self):  # pragma: no cover
         """Implemented by child class"""

ansys/mapdl/core/mapdl_grpc.py

@@ -2004,4 +2004,4 @@ def erinqr(self, key, **kwargs):
     def wrinqr(self, key, **kwargs):
         """Wrap the ``wrinqr`` method to take advantage of the gRPC methods."""
         super().wrinqr(key, pname=TMP_VAR, mute=True, **kwargs)
-        return self.scalar_param(TMP_VAR)
+        return self.scalar_param(TMP_VAR)

tests/conftest.py

@@ -17,7 +17,6 @@
 )
 from common import get_details_of_nodes, get_details_of_elements, Node, Element
 
-
 # Necessary for CI plotting
 pyvista.OFF_SCREEN = True