refactor: move hpcname & scontrol to hpc module

src/ccbr_tools/pipeline/hpc.py

@@ -5,8 +5,8 @@
 which contains default attributes for supported clusters.
 """
 
-from .util import get_hpcname
 from .cache import get_singularity_cachedir
+from ..shell import shell_run
 
 
 class Cluster:
@@ -53,8 +53,12 @@ def __init__(self):
         super().__init__()
         self.name = "biowulf"
         self.modules = {
-            "nxf": "ccbrpipeliner nextflow",
-            "smk": "ccbrpipeliner snakemake/7 singularity",
+            "nxf": "nextflow" + ""
+            if is_loaded(module="ccbrpipeliner")
+            else " ccbrpipeliner",
+            "smk": "snakemake/7 singularity" + ""
+            if is_loaded(module="ccbrpipeliner")
+            else " ccbrpipeliner",
         }
         self.singularity_sif_dir = "/data/CCBR_Pipeliner/SIFs"
 
@@ -102,3 +106,47 @@ def get_hpc(debug=False):
     hpc_options = {"biowulf": Biowulf, "frce": FRCE}
     hpc_name = get_hpcname() if not debug else debug
     return hpc_options.get(hpc_name, Cluster)()
+
+
+def get_hpcname():
+    """
+    Get the HPC name using scontrol
+
+    Returns:
+        hpcname (str): The HPC name  (biowulf, frce, or an empty string)
+    """
+    scontrol_out = scontrol_show()
+    hpc = scontrol_out["ClusterName"] if "ClusterName" in scontrol_out.keys() else ""
+    if hpc == "fnlcr":
+        hpc = "frce"
+    return hpc
+
+
+def scontrol_show():
+    """
+    Run `scontrol show config` and parse the output as a dictionary
+
+    Returns:
+        scontrol_dict (dict): dictionary containing the output of `scontrol show config`
+    """
+    scontrol_dict = dict()
+    scontrol_out = shell_run(
+        "scontrol show config", shell=True, capture_output=True, text=True
+    )
+    if len(scontrol_out) > 0:
+        for line in scontrol_out.split("\n"):
+            line_split = line.split("=")
+            if len(line_split) > 1:
+                scontrol_dict[line_split[0].strip()] = line_split[1].strip()
+    return scontrol_dict
+
+
+def is_loaded(module="ccbrpipeliner"):
+    """
+    Check whether the ccbrpipeliner module is loaded
+
+    Returns:
+        is_loaded (bool): True if the module is loaded, False otherwise
+    """
+    output = shell_run("bash -c 'module list'")
+    return module in output

src/ccbr_tools/pipeline/nextflow.py

@@ -34,7 +34,6 @@ def run(
     nextflow_command = ["nextflow", "run", nextfile_path]
 
     hpc = get_hpc()
-    print("HPC", hpc.name)
     if mode == "slurm" and not hpc:
         raise ValueError("mode is 'slurm' but no HPC environment was detected")
     # add any additional Nextflow commands

src/ccbr_tools/pipeline/util.py

@@ -17,39 +17,7 @@
 import yaml
 
 from ..pkg_util import repo_base, msg
-
-
-def scontrol_show():
-    """
-    Run `scontrol show config` and parse the output as a dictionary
-
-    Returns:
-        scontrol_dict (dict): dictionary containing the output of `scontrol show config`
-    """
-    scontrol_dict = dict()
-    scontrol_out = subprocess.run(
-        "scontrol show config", shell=True, capture_output=True, text=True
-    ).stdout
-    if len(scontrol_out) > 0:
-        for line in scontrol_out.split("\n"):
-            line_split = line.split("=")
-            if len(line_split) > 1:
-                scontrol_dict[line_split[0].strip()] = line_split[1].strip()
-    return scontrol_dict
-
-
-def get_hpcname():
-    """
-    Get the HPC name using scontrol
-
-    Returns:
-        hpcname (str): The HPC name  (biowulf, frce, or an empty string)
-    """
-    scontrol_out = scontrol_show()
-    hpc = scontrol_out["ClusterName"] if "ClusterName" in scontrol_out.keys() else ""
-    if hpc == "fnlcr":
-        hpc = "frce"
-    return hpc
+from .hpc import get_hpcname
 
 
 def get_tmp_dir(tmp_dir, outdir, hpc=get_hpcname()):

tests/test_cli.py

@@ -1,5 +1,5 @@
 from ccbr_tools.shell import shell_run
-from ccbr_tools.pipeline.util import get_hpcname
+from ccbr_tools.pipeline.hpc import get_hpcname
 
 
 def test_version():

tests/test_hpc.py

@@ -1,4 +1,6 @@
 from ccbr_tools.pipeline.hpc import get_hpc
+from ccbr_tools.shell import shell_run
+import subprocess
 
 
 def test_hpc_biowulf():
@@ -29,4 +31,5 @@ def test_hpc_none():
     assert not any([hpc, hpc.name, *hpc.modules.values(), hpc.singularity_sif_dir])
 
 
-print(get_hpc())
+if __name__ == "__main__":
+    print(subprocess.run("bash -c 'module list'", shell=True))