tools: Add function for parsing schism version

Related to ec-jrc#125

pyposeidon/tools.py

@@ -8,19 +8,23 @@
 import os
 import shlex
 import pathlib
+import re
 import subprocess
 from collections.abc import Iterable
 import time
 import shutil
 
 import psutil
 import xarray as xr
-import pandas as pd
 import numpy as np
 import rioxarray
 
 logger = logging.getLogger(__name__)
 
+
+SCHISM_VERSION_PATTERN = re.compile(r"schism v(\d+\.\d+\.\d+)\w*")
+
+
 LAUNCH_SCHISM_TEMPLATE = """
 #!/usr/bin/env bash
 #
@@ -70,6 +74,28 @@
 """.strip()
 
 
+# TODO Handle master/develop version
+def parse_schism_version(version_output: str) -> str:
+    try:
+        version_line = version_output.strip().splitlines()[0]
+        version = SCHISM_VERSION_PATTERN.match(version_line).group(1)
+        return version
+    except Exception as exc:
+        raise ValueError(f"Failed to parse version from:\n {version_output}") from exc
+
+
+def get_schism_version() -> str:
+    cmd = "schism -v"
+    proc = subprocess.run(
+        shlex.split(cmd),
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+    version = parse_schism_version(proc.stdout)
+    return version
+
+
 def get_solver(solver_name: str):
     # Panos: We do the imports here, because there is some sort of cyclical imports issue
     # and no time to properly solve it

tests/test_tools.py

@@ -11,6 +11,25 @@
 from . import DATA_DIR
 
 
+@pytest.mark.parametrize(
+    "version_output,expected_version",
+    [
+        ("schism v5.9.0mod", "5.9.0"),
+        ("schism v5.9.0", "5.9.0"),
+        ("schism v5.10.1", "5.10.1"),
+    ],
+)
+def test_parse_schism_version(version_output, expected_version):
+    assert tools.parse_schism_version(version_output) == expected_version
+
+
+@pytest.mark.skipif(not shutil.which("schism"), reason="requires schism binary")
+def test_get_schism_version():
+    version = tools.get_schism_version()
+    assert isinstance(version, str)
+    assert len(version) > 0
+
+
 @pytest.mark.skipif(not shutil.which("mpirun"), reason="requires MPI backend")
 @pytest.mark.parametrize("use_threads", [True, False])
 @pytest.mark.parametrize("scribes", [-1, 1])