Source code for ewoksmx.shell_utils.execute_results

import logging
import os
import pathlib
import sys
from typing import Optional

_logger = logging.getLogger("__name___")




class BashExecutionResult:
    """Bash execution return code and log files."""

    def __init__(
        self,
        script_path: pathlib.Path,
        stdout: bool = True,
        stderr: bool = True,
        stdmerge: bool = True,
    ):
        self.script_path = script_path

        self.log_path = None
        self.stdout_path = None
        self.stderr_path = None

        log_name = script_path.stem
        log_directory = script_path.parent

        if stdout and stderr:
            if stdmerge:
                self.log_path = log_directory / f"{log_name}.log"
            else:
                self.stdout_path = log_directory / f"{log_name}.stdout"
                self.stderr_path = log_directory / f"{log_name}.stderr"
        elif stdout:
            self.stdout_path = log_directory / f"{log_name}.stdout"
        elif stderr:
            self.stderr_path = log_directory / f"{log_name}.stderr"

        self.return_code = None
        self.slurm_id = None



    def read_stdout(self) -> Optional[str]:
        return self._read_file(self.stdout_path)




    def read_stderr(self) -> Optional[str]:
        return self._read_file(self.stderr_path)




    def read_log(self) -> Optional[str]:
        return self._read_file(self.log_path)




    def raise_on_error(self):
        if self.return_code is not None and self.return_code != 0:
            if self.log_path:
                suffix = f" Check {self.log_path} for details."
            elif self.stderr_path:
                suffix = f" Check {self.stderr_path} for details."
            elif self.stdout_path:
                suffix = f" Check {self.stdout_path} for details."
            else:
                suffix = ""
            raise RuntimeError(
                f"Shell script {self.script_path} failed with return code {self.return_code}.{suffix}"
            )


    def _read_file(self, file_path: Optional[str]) -> Optional[str]:
        if file_path and os.path.exists(file_path):
            with open(file_path, "r") as f:
                return f.read()



    def log(self, level: int = logging.DEBUG):
        if self.log_path:
            logs = self.read_log()
        else:
            out = self.read_stdout()
            err = self.read_stderr()
            if out and err:
                logs = f"=== STDOUT ====\n{out}\n=== STDERR ====\n{err}"
            else:
                logs = out or err

        if logs:
            _logger.log(level, logs)




    def print(self):
        if self.log_path:
            logs = self.read_log()
            print(logs)
        else:
            out = self.read_stdout()
            err = self.read_stderr()
            if out:
                print(out, file=sys.stdout)
            if err:
                print(out, file=sys.stderr)