Source code for dls_utilpack.bash_composer

import os
import stat
from typing import List, Optional


# ---------------------------------------------------------------------
[docs]class Element: """ Base class for other elements. """ def compose(self) -> List[str]: return []
# --------------------------------------------------------------------- class Prolog(Element): def __init__(self): pass def compose(self) -> List[str]: lines = [] lines.append("#!/bin/bash") lines.append("") lines.append("# -----------------------------------------------------") lines.append("function __execute {") lines.append(" echo '$' $1") lines.append(" eval $1") lines.append(" local rc=$?") lines.append(" if [ $rc -ne $2 ]") lines.append(" then") lines.append(" echo 'command failed with rc' $rc") lines.append(" exit 1") lines.append(" fi") lines.append("}") lines.append("# -----------------------------------------------------") return lines # --------------------------------------------------------------------- class Command(Element): def __init__(self, command: str, expected_rc: Optional[int] = 0): self.__command = command self.__expected_rc = expected_rc def compose(self) -> List[str]: lines = [] lines.append("") lines.append(f"__execute '{self.__command}' {self.__expected_rc}") return lines # --------------------------------------------------------------------- class Raw(Element): def __init__(self, raw_line): self.__raw_line = raw_line def compose(self) -> List[str]: lines = [] lines.append(self.__raw_line) return lines # --------------------------------------------------------------------- class Print(Element): def __init__(self, message): self.__message = message def compose(self) -> List[str]: lines = [] lines.append("") lines.append(f"eval echo '{self.__message}'") return lines # --------------------------------------------------------------------- class LoadModules(Element): def __init__(self, directories: List[str], modules: List[str]): """ Create object which composed to load linux environment modules in the bash script. Args: directories: List of directories for the "module use" command. modules: List of names for the "module load" command. Returns: List[str]: Lines to be added to the bash script. """ self.__directories = directories self.__modules = modules def compose(self) -> List[str]: composer = BashComposer(should_include_prolog=False) # Establish a USER environment variable. composer.add(Raw('export USER="`/usr/bin/id -un`"')) # Establish access to linux environment modules. MODULESHOME = os.environ.get("MODULESHOME") if MODULESHOME is None: raise RuntimeError("MODULESHOME environment variable is not set") composer.add(Command(f"source {MODULESHOME}/init/bash")) # Load the module which supports the ptyrex_recon script. composer.add(Command("module purge")) for directory in self.__directories: composer.add(Command(f"module use --append {directory}")) for module in self.__modules: composer.add(Command(f"module load {module} 2>&1")) # Put a few things in the log. composer.add(Command("module list 2>&1")) composer.add(Command("python3 --version")) composer.add(Print("------ end of modules_load_bash_lines ------")) return composer.compose_lines() # ---------------------------------------------------------------------
[docs]class BashComposer: """ Class which helps with composing bash scripts consistently and reliably. """ def __init__(self, should_include_prolog: Optional[bool] = True): """ Args: should_include_prolog: True if should emit a prolog on this bash script. Defaults to True. """ self.__elements: List[Element] = [] if should_include_prolog: self.add(Prolog()) # -----------------------------------------------------------------
[docs] def add(self, element: Element) -> None: """ Add element to be included in the bash script. Possible element classes are: Print, Raw, Command, Prolog and LoadModules. Args: element (Element): Element object. """ self.__elements.append(element)
# -----------------------------------------------------------------
[docs] def add_print(self, message: str) -> None: """ Add print element to bash script. Args: message (str): Message to be printed. """ self.add(Print(message))
# ----------------------------------------------------------------- def add_command(self, command: str, expected_rc: Optional[int] = 0) -> None: self.add(Command(command, expected_rc)) # -----------------------------------------------------------------
[docs] def add_load_modules(self, directories: List[str], modules: List[str]) -> None: """ Add shell commands for loading linux environment modules. Args: directories: List of directories for the "module use" command. modules: List of names for the "module load" command. """ self.add(LoadModules(directories, modules))
# -----------------------------------------------------------------
[docs] def compose_lines(self) -> List[str]: """ Return bash script as list of lines. Returns: The complete bash script. """ lines = [] for element in self.__elements: lines.extend(element.compose()) return lines
# -----------------------------------------------------------------
[docs] def compose_string(self) -> str: """ Return composed bash script as a string. Returns: str: The complete bash script. """ lines = [] for element in self.__elements: lines.extend(element.compose()) return "\n".join(lines)
# -----------------------------------------------------------------
[docs] def write(self, filename: str) -> None: """ Write composed bash script lines to file. Adds execution permission to the file. """ with open(filename, "w") as stream: stream.write(self.compose_string()) stream.write("\n") st = os.stat(filename) os.chmod(filename, st.st_mode | stat.S_IEXEC)