Safer cpac call generation; Refactoring

src/ecpac/bash_builder.py

@@ -0,0 +1,134 @@
+"""Functions for building bash scripts."""
+
+import datetime
+import shlex
+from os import PathLike
+
+from ecpac import utils
+
+SCHEBANG = "#!/usr/bin/bash"
+
+
+def _buf_pad_before_if_not_empty(buf: list[str], before: str = "") -> list[str]:
+    """Pad a buffer with a string if the buffer is not empty."""
+    if len(buf) > 0:
+        return [before, *buf]
+    return buf
+
+
+def _buf_pad_after_if_not_empty(buf: list[str], after: str = "") -> list[str]:
+    """Pad a buffer with a string if the buffer is not empty."""
+    if len(buf) > 0:
+        return [*buf, after]
+    return buf
+
+
+def _sbatch_header(**kwargs) -> list[str]:  # noqa: ANN003
+    """Generate the header for a sbatch job script.
+
+    _ in keywords are replaced with -.
+    """
+    return [f"#SBATCH --{k.replace('_', '-')} {v}" for k, v in kwargs.items()]
+
+
+def job_template(  # noqa: PLR0913
+    *,
+    job_name: str,
+    job_stdout_file: str,
+    job_duration_limit: datetime.timedelta,
+    job_threads: int,
+    job_memory_gb: float,
+    wd: str | PathLike,
+    path_input: str | PathLike,
+    path_output: str | PathLike,
+    cpac_threads: int,
+    cpac_memory_gb: float,
+    path_image: str | PathLike,
+    analysis_level: str,
+    subject: str,
+    pipeline: str,
+    pipeline_is_preconfig: bool,
+    cpac_sources: str | PathLike | None = None,
+    extra_cpac_args: str | None = None,
+    before_run: list[str] | None = None,
+    after_run: list[str] | None = None,
+) -> str:
+    """Generate a bash script for running C-PAC with sbatch."""
+    before_run = before_run if before_run is not None else []
+    after_run = after_run if after_run is not None else []
+
+    cpac_call = [
+        "singularity",
+        "run",
+        "--cleanenv",
+    ]
+
+    # Patch C-PAC sources if available
+    if cpac_sources:
+        cpac_call.extend(
+            [
+                "-B",
+                f"{cpac_sources}/CPAC:/code/CPAC",
+                "-B",
+                f"{cpac_sources}/dev/docker_data/run.py:/code/run.py",
+                "-B",
+                f"{cpac_sources}/dev/docker_data:/cpac_resources",
+            ]
+        )
+
+    cpac_call.extend(
+        [
+            "-B",
+            f"{path_input}:{path_input}:ro",
+            "-B",
+            f"{path_output}:{path_output}",
+        ]
+    )
+
+    pipeline_args = ["--preconfig" if pipeline_is_preconfig else "--pipeline-file", str(pipeline)]
+
+    cpac_call.extend(
+        [
+            str(path_image),
+            str(path_input),
+            str(path_output),
+            analysis_level,
+            "--skip_bids_validator",
+            "--n_cpus",
+            str(cpac_threads),
+            "--mem_gb",
+            str(cpac_memory_gb),
+            "--participant_label",
+            subject,
+            *pipeline_args,
+        ]
+    )
+
+    cpac_call_str = shlex.join(cpac_call)
+
+    if extra_cpac_args:
+        cpac_call_str += f" {extra_cpac_args}"
+
+    buf = [
+        SCHEBANG,
+        *_sbatch_header(
+            job_name=job_name,
+            output=job_stdout_file,
+            nodes=1,
+            partition="RM-shared",
+            time=utils.timedelta_to_hms(job_duration_limit),
+            ntasks_per_node=job_threads,
+            mem=utils.bridges_gb_to_mb(job_memory_gb),
+        ),
+        "",
+        "set -x",
+        "",
+        shlex.join(["cd", str(wd)]) + " || exit",
+        "",
+        *_buf_pad_after_if_not_empty(before_run),
+        cpac_call_str,
+        *_buf_pad_before_if_not_empty(after_run),
+        "",
+    ]
+
+    return "\n".join(buf)

src/ecpac/cli.py

@@ -33,3 +33,13 @@ def option_or_confirm(opt: bool | None, prompt: str, *, default: bool | None = F
     if opt is not None:
         return opt
     return click.confirm(prompt, default=default)
+
+
+def icon_message(icon: str, text: str) -> str:
+    """Styled message with a prefixed icon."""
+    return icon + " " + click.style(text, fg="blue")
+
+
+def icon_message_emph(icon: str, text: str) -> str:
+    """Styled message with a prefixed icon and emphasis."""
+    return icon + " " + click.style(text, bg="blue")

src/ecpac/consts.py

@@ -33,42 +33,3 @@
     "rbc-options",
     "rodent",
 ]
-
-
-BASH_TEMPLATE_JOB = """\
-#!/usr/bin/bash
-#SBATCH --job-name {job_name}
-#SBATCH --output {stdout_file}
-#SBATCH --nodes 1
-#SBATCH --partition RM-shared
-#SBATCH --time {duration_str}
-#SBATCH --ntasks-per-node {threads}
-#SBATCH --mem {memory_mb}
-
-set -x
-
-cd {wd}
-
-{before_run}singularity run \
---cleanenv \
-{cpac_bin_opt} \
--B {path_input}:{path_input}:ro \
--B {path_output}:{path_output} \
-{image} {path_input} {path_output} {analysis_level} \
---skip_bids_validator \
---n_cpus {cpac_threads} \
---mem_gb {cpac_memory_gb} \
---participant_label {subject} \
-{pipeline} \
-{extra_cpac_args}{after_run}
-"""
-
-BASH_TEMPLATE_PIPELINE_PRECONFIG = "--preconfig {pipeline}"
-BASH_TEMPLATE_PIPELINE_CONFIG_FILE = "--pipeline-file {pipeline}"
-
-
-BASH_TEMPLATE_JOB_CPAC_BIN = """\
--B {cpac_bin}/CPAC:/code/CPAC \
--B {cpac_bin}/dev/docker_data/run.py:/code/run.py \
--B {cpac_bin}/dev/docker_data:/cpac_resources \
-"""

src/ecpac/fsplan.py

@@ -0,0 +1,28 @@
+"""File system plan. Used to preview and apply changes to the file system."""
+
+import dataclasses
+import pathlib as pl
+import stat
+
+
+@dataclasses.dataclass
+class FsPlan:
+    """Planned file system change."""
+
+    path: pl.Path
+    is_file: bool = False
+    contents_text: str | None = None
+    make_executable: bool = False
+
+    def apply(self) -> None:
+        """Apply the file system change."""
+        if self.is_file:
+            self.path.parent.mkdir(parents=True, exist_ok=True)
+            txt = "" if self.contents_text is None else self.contents_text
+            with self.path.open("w", encoding="utf-8") as handle:
+                handle.write(txt)
+
+            if self.make_executable:
+                self.path.chmod(self.path.stat().st_mode | stat.S_IEXEC)
+        else:
+            self.path.mkdir(parents=True, exist_ok=True)