Feat/legacy caseread

.github/workflows/ci.yml

@@ -22,7 +22,7 @@ env:
   # you might end up reusing a previous cache if it haven't been deleted already.
   # It applies 7 days retention policy by default.
   RESET_EXAMPLES_CACHE: 5
-  API_CODE_CACHE: 1
+  API_CODE_CACHE: 2
 
 jobs:
   stylecheck:

src/ansys/fluent/core/filereader/casereader.py

@@ -14,7 +14,8 @@
 input_parameters = reader.input_parameters()
 output_parameters = reader.output_parameters()
 """
-
+import codecs
+import gzip
 from pathlib import Path
 from typing import List
 
@@ -56,6 +57,8 @@ def __init__(self, raw_data):
         for elem in parameter:
             if len(elem) and elem[0] == "name":
                 self.name = elem[1][1]
+            if len(elem) and elem[0] == "fluent-units":
+                self.units = elem[1].strip()
 
 
 class CaseReader:
@@ -73,36 +76,53 @@ class CaseReader:
         Get the precision (1 or 2 for 1D of 2D)
     """
 
-    def __init__(self, hdf5_case_filepath: str):
+    def __init__(self, case_filepath: str):
+
         try:
-            file = h5py.File(hdf5_case_filepath)
+            if "".join(Path(case_filepath).suffixes) == ".cas.h5":
+                file = h5py.File(case_filepath)
+                settings = file["settings"]
+                rpvars = settings["Rampant Variables"][0]
+                rp_vars_str = rpvars.decode()
+            elif Path(case_filepath).suffix == ".cas":
+                with open(case_filepath, "rb") as file:
+                    rp_vars_str = file.read()
+                rp_vars_str = _get_processed_string(rp_vars_str)
+            elif "".join(Path(case_filepath).suffixes) == ".cas.gz":
+                with gzip.open(case_filepath, "rb") as file:
+                    rp_vars_str = file.read()
+                rp_vars_str = _get_processed_string(rp_vars_str)
+            else:
+                raise RuntimeError()
+
         except FileNotFoundError:
-            raise RuntimeError(f"The case file {hdf5_case_filepath} cannot be found.")
+            raise RuntimeError(f"The case file {case_filepath} cannot be found.")
+
         except OSError:
             error_message = (
-                "Could not read case file. " "Only valid HDF5 files can be read. "
+                "Could not read case file. "
+                "Only valid Case files (.h5, .cas, .cas.gz) can be read. "
             )
-            if Path(hdf5_case_filepath).suffix != ".h5":
-                error_message += (
-                    f"The file {hdf5_case_filepath} does not have a .h5 extension."
-                )
             raise RuntimeError(error_message)
+
         except BaseException:
-            raise RuntimeError(f"Could not read case file {hdf5_case_filepath}")
-        settings = file["settings"]
-        rpvars = settings["Rampant Variables"][0]
-        rp_vars_str = rpvars.decode()
+            raise RuntimeError(f"Could not read case file {case_filepath}")
+
         self._rp_vars = lispy.parse(rp_vars_str)[1]
         self._rp_var_cache = {}
 
     def input_parameters(self) -> List[InputParameter]:
         exprs = self._named_expressions()
-        input_params = []
-        for expr in exprs:
-            for attr in expr:
-                if attr[0] == "input-parameter" and attr[1] is True:
-                    input_params.append(InputParameter(expr))
-        return input_params
+        if exprs:
+            input_params = []
+            for expr in exprs:
+                for attr in expr:
+                    if attr[0] == "input-parameter" and attr[1] is True:
+                        input_params.append(InputParameter(expr))
+            return input_params
+        else:
+            parameters = self._find_rp_var("parameters/input-parameters")
+            return [InputParameter(param) for param in parameters]
 
     def output_parameters(self) -> List[OutputParameter]:
         parameters = self._find_rp_var("parameters/output-parameters")
@@ -132,3 +152,21 @@ def _find_rp_var(self, name: str):
                 if type(var) == list and len(var) and var[0] == name:
                     self._rp_var_cache[name] = var[1]
                     return var[1]
+
+
+def _get_processed_string(input_string: bytes) -> str:
+    """Processes the input string (binary) with help of an identifier to return
+    it in a format which can be parsed by lispy.parse()
+
+    Parameters
+    ----------
+    input_string : bytes
+        The input string in bytes
+
+    Returns
+    -------
+    processed string (str)
+    """
+    rp_vars_str = codecs.decode(input_string, errors="ignore")
+    string_identifier = "(37 ("
+    return string_identifier + rp_vars_str.split(string_identifier)[1]

tests/test_casereader.py

@@ -1,14 +1,10 @@
 from ansys.fluent.core import examples
-from ansys.fluent.core.filereader.casereader import CaseReader
+from ansys.fluent.core.filereader.casereader import CaseReader, _get_processed_string
 
 
-def test_casereader():
+def call_casereader(case_filepath: str):
 
-    case_filepath = examples.download_file(
-        "Static_Mixer_Parameters.cas.h5", "pyfluent/static_mixer"
-    )
-
-    reader = CaseReader(hdf5_case_filepath=case_filepath)
+    reader = CaseReader(case_filepath=case_filepath)
 
     input_parameters = reader.input_parameters()
 
@@ -32,15 +28,64 @@ def test_casereader():
 
     assert len(output_parameters) == 2
 
-    assert {"outlet-temp-avg-op", "outlet-vel-avg-op"} == {
-        p.name for p in output_parameters
-    }
+    output_parameter_dict = {p.name: p.units for p in output_parameters}
+    assert {
+        "outlet-temp-avg-op": "K",
+        "outlet-vel-avg-op": "m s^-1",
+    } == output_parameter_dict
+
+
+def test_casereader_h5():
+    call_casereader(
+        examples.download_file(
+            "Static_Mixer_Parameters.cas.h5", "pyfluent/static_mixer"
+        )
+    )
+
+
+def test_casereader_binary_cas():
+    call_casereader(
+        examples.download_file(
+            "Static_Mixer_Parameters_legacy_binary.cas", "pyfluent/static_mixer"
+        )
+    )
+
+
+def test_casereader_binary_gz():
+    call_casereader(
+        examples.download_file(
+            "Static_Mixer_Parameters_legacy_binary.cas.gz", "pyfluent/static_mixer"
+        )
+    )
+
+
+def test_casereader_text_cas():
+    call_casereader(
+        examples.download_file(
+            "Static_Mixer_Parameters_legacy_text.cas", "pyfluent/static_mixer"
+        )
+    )
+
+
+def test_casereader_text_gz():
+    call_casereader(
+        examples.download_file(
+            "Static_Mixer_Parameters_legacy_text.cas.gz", "pyfluent/static_mixer"
+        )
+    )
+
+
+def test_processed_string():
+    assert (
+        _get_processed_string(b"Hello! World (37 ( Get this part of the string ))")
+        == "(37 ( Get this part of the string ))"
+    )
 
 
 def test_casereader_no_file():
     throws = False
     try:
-        reader = CaseReader(hdf5_case_filepath="no_file.cas.h5")
-    except BaseException:
+        call_casereader("no_file.cas.h5")
+    except RuntimeError:
         throws = True
     assert throws