PrefixCommand

Description

General utilities for prefix commands (try, parallel, submit).
The explicit options (`cmds`) are unchanged; among the implicit options,
the inputs are checked (if any), and the output is adapted to each thread.

Public Member Functions
None	__init__ (self, str prefix, List[str] cmds)
None	parse (self, List[str] args)
int	get_entries (self, input)
None	prepare_io (self, bool multi=True)
None	prepare_fire_and_forget (self)
None	clean_up_last_run (self)
None	__init__ (self, str prefix, List[str] cmds)
None	parse (self, List[str] args)
int	get_entries (self, input)
None	prepare_io (self, bool multi=True)
None	prepare_fire_and_forget (self)
None	clean_up_last_run (self)

Public Attributes
	name
	cmds
	helper
	splittable
	nSplit
	extension
	inputs
	output
	args
	absinputs
	absoutput

Static Public Attributes
	str
	Path

Private Member Functions
str	_help_text (self)
List[str]	_synopsis (self)
None	_prepare_single_output (self)
None	_prepare_multi_output (self)
None	_prepare_input (self)
None	_save_environment (self)
str	_help_text (self)
List[str]	_synopsis (self)
None	_prepare_single_output (self)
None	_prepare_multi_output (self)
None	_prepare_input (self)
None	_save_environment (self)

Constructor & Destructor Documentation

__init__() [1/2]

None __init__	(		self,
		str	prefix,
		List[str]	cmds
	)

Constructor from a separated list of arguments.

    def __init__(self, prefix: str, cmds: List[str]) -> None:
        """Constructor from a separated list of arguments."""
 
        self.name = os.path.basename(prefix)
        self.cmds = cmds
 
        self.helper = self._help_text()
        if self.helper[0].count("output") != 1:
            raise ValueError(
                f"`{self.cmds.exec}`"
                + " does not match the requirements to be run with this prefix command: "
                + "there must be exactly one `output` argument."
            )
 
        self.splittable = ["nSplit" in row for row in self.helper[1:]].count(True)
        self.nSplit = cmds.nSplit if self.splittable else 1
 
        self.extension = (
            check_output([self.cmds.exec] + ["-o"])
            .decode()
            .strip()
            .split("\n")[0]
            .split(" ")[0]
        )
 

__init__() [2/2]

None __init__	(		self,
		str	prefix,
		List[str]	cmds
	)

Constructor from a separated list of arguments.

    def __init__(self, prefix: str, cmds: List[str]) -> None:
        """Constructor from a separated list of arguments."""
 
        self.name = os.path.basename(prefix)
        self.cmds = cmds
 
        self.helper = self._help_text()
        if self.helper[0].count("output") != 1:
            raise ValueError(
                f"`{self.cmds.exec}`"
                + " does not match the requirements to be run with this prefix command: "
                + "there must be exactly one `output` argument."
            )
 
        self.splittable = ["nSplit" in row for row in self.helper[1:]].count(True)
        self.nSplit = cmds.nSplit if self.splittable else 1
 
        self.extension = (
            check_output([self.cmds.exec] + ["-o"])
            .decode()
            .strip()
            .split("\n")[0]
            .split(" ")[0]
        )
 

Member Function Documentation

_help_text() [1/2]

str _help_text (   self )

private

Returns the command's normal help text.

    def _help_text(self) -> str:
        """Returns the command's normal help text."""
        # if executable is given, then the helper is adapted to match with the prefix command
        text = (
            check_output(self.cmds.exec + " -h", shell=True)
            .decode()
            .strip()
            .split("\n")
        )
        return text
 

_help_text() [2/2]

str _help_text (   self )

private

Returns the command's normal help text.

    def _help_text(self) -> str:
        """Returns the command's normal help text."""
        # if executable is given, then the helper is adapted to match with the prefix command
        text = (
            check_output(self.cmds.exec + " -h", shell=True)
            .decode()
            .strip()
            .split("\n")
        )
        return text
 

_prepare_input() [1/2]

None _prepare_input (   self )

private

Resolve the paths to the input(s).

    def _prepare_input(self) -> None:
        """Resolve the paths to the input(s)."""
 
        for x in self.inputs:
            self.absinputs += glob(x)
        if len(self.inputs) > 0 and len(self.absinputs) == 0:
            raise ValueError("Inputs could not be found: " + " ".join(self.inputs))
 
        self.absinputs = [str(Path(x).resolve()) for x in self.absinputs]
 

_prepare_input() [2/2]

None _prepare_input (   self )

private

Resolve the paths to the input(s).

    def _prepare_input(self) -> None:
        """Resolve the paths to the input(s)."""
 
        for x in self.inputs:
            self.absinputs += glob(x)
        if len(self.inputs) > 0 and len(self.absinputs) == 0:
            raise ValueError("Inputs could not be found: " + " ".join(self.inputs))
 
        self.absinputs = [str(Path(x).resolve()) for x in self.absinputs]
 

_prepare_multi_output() [1/2]

None _prepare_multi_output (   self )

private

Prepare the members for parallel running (local or on farm).

    def _prepare_multi_output(self) -> None:
        """Prepare the members for parallel running (local or on farm)."""
 
        if self.absoutput.suffix != "":
            raise ValueError(
                f"A directory with an extension is misleading: {self.output}"
            )
 
        if self.absoutput.exists():
            if self.absoutput.is_file():
                raise ValueError(self.output + " is not a directory")
 
            new_rootfiles = []
            for k in range(self.nSplit):
                rootfile = self.absoutput / f"{k}{self.extension}"
                if rootfile.exists() and not os.access(rootfile, os.W_OK):
                    raise ValueError(
                        "The root files in " + self.output + " are not writable"
                    )
                new_rootfiles += [str(rootfile)]
 
            existing_rootfiles = glob(str(self.absoutput / f"*{self.extension}"))
            if len(set(existing_rootfiles).difference(new_rootfiles)) > 0:
                raise ValueError(
                    "Residual files from a former run will not be overwritten."
                )
        else:
            self.absoutput.mkdir(parents=True)
 
        self._save_environment()
 

_prepare_multi_output() [2/2]

None _prepare_multi_output (   self )

private

Prepare the members for parallel running (local or on farm).

    def _prepare_multi_output(self) -> None:
        """Prepare the members for parallel running (local or on farm)."""
 
        if self.absoutput.suffix != "":
            raise ValueError(
                f"A directory with an extension is misleading: {self.output}"
            )
 
        if self.absoutput.exists():
            if self.absoutput.is_file():
                raise ValueError(self.output + " is not a directory")
 
            new_rootfiles = []
            for k in range(self.nSplit):
                rootfile = self.absoutput / f"{k}{self.extension}"
                if rootfile.exists() and not os.access(rootfile, os.W_OK):
                    raise ValueError(
                        "The root files in " + self.output + " are not writable"
                    )
                new_rootfiles += [str(rootfile)]
 
            existing_rootfiles = glob(str(self.absoutput / f"*{self.extension}"))
            if len(set(existing_rootfiles).difference(new_rootfiles)) > 0:
                raise ValueError(
                    "Residual files from a former run will not be overwritten."
                )
        else:
            self.absoutput.mkdir(parents=True)
 
        self._save_environment()
 

_prepare_single_output() [1/2]

None _prepare_single_output (   self )

private

Prepare the working area for a single-run command.

    def _prepare_single_output(self) -> None:
        """Prepare the working area for a single-run command."""
 
        if self.absoutput.exists():
            if not os.access(self.absoutput, os.W_OK):
                raise ValueError(self.absoutput + " is not writable")
            if self.absoutput.is_dir():
                self.absoutput /= "0" + self.extension
        else:
            if len(self.absoutput.suffix) == 0:  # then assume it's a directory
                self.absoutput.mkdir(parents=True)
                self.absoutput /= "0" + self.extension
 
        if self.absoutput.is_dir():
            self._save_environment()
 

_prepare_single_output() [2/2]

None _prepare_single_output (   self )

private

Prepare the working area for a single-run command.

    def _prepare_single_output(self) -> None:
        """Prepare the working area for a single-run command."""
 
        if self.absoutput.exists():
            if not os.access(self.absoutput, os.W_OK):
                raise ValueError(self.absoutput + " is not writable")
            if self.absoutput.is_dir():
                self.absoutput /= "0" + self.extension
        else:
            if len(self.absoutput.suffix) == 0:  # then assume it's a directory
                self.absoutput.mkdir(parents=True)
                self.absoutput /= "0" + self.extension
 
        if self.absoutput.is_dir():
            self._save_environment()
 

_save_environment() [1/2]

None _save_environment (   self )

private

Save environment in a file.

    def _save_environment(self) -> None:
        """Save environment in a file."""
 
        env = self.absoutput / ".env"
        with open(env, "w") as f:
            for key, value in os.environ.items():
                if key in ["PWD", "OLDPWD", "TMPDIR", "KRB5CCNAME", "EOS_MGM_URL"]:
                    continue
                if "BASH" in key:
                    continue
                value = shlex.quote(value)
                print(f"export {key}={value}", file=f)
        os.chmod(env, stat.S_IRWXU)
 

_save_environment() [2/2]

None _save_environment (   self )

private

Save environment in a file.

    def _save_environment(self) -> None:
        """Save environment in a file."""
 
        env = self.absoutput / ".env"
        with open(env, "w") as f:
            for key, value in os.environ.items():
                if key in ["PWD", "OLDPWD", "TMPDIR", "KRB5CCNAME", "EOS_MGM_URL"]:
                    continue
                if "BASH" in key:
                    continue
                value = shlex.quote(value)
                print(f"export {key}={value}", file=f)
        os.chmod(env, stat.S_IRWXU)
 

_synopsis() [1/2]

List[str] _synopsis (   self )

private

Extract synopsis from helper (after undoing the bold characters).

    def _synopsis(self) -> List[str]:
        """Extract synopsis from helper (after undoing the bold characters)."""
        synopsis = self.helper[0].replace("\x1b[1m", "").replace("\x1b[0m", "")
        return synopsis.split()[1:]
 

_synopsis() [2/2]

List[str] _synopsis (   self )

private

Extract synopsis from helper (after undoing the bold characters).

    def _synopsis(self) -> List[str]:
        """Extract synopsis from helper (after undoing the bold characters)."""
        synopsis = self.helper[0].replace("\x1b[1m", "").replace("\x1b[0m", "")
        return synopsis.split()[1:]
 

clean_up_last_run() [1/2]

None clean_up_last_run

(

self

)

Removes the executable, libraries, hidden files, and dictionaries from the output directory.

    def clean_up_last_run(self) -> None:
        """Removes the executable, libraries, hidden files, and dictionaries from the output directory."""
 
        exec = Path(which(self.cmds.exec)).name
 
        for file in [exec, ".env", ".condor"]:
            file = self.absoutput / Path(file)
            if file.is_file():
                os.remove(str(file))
 
        for ext in ["so", "pcm"]:
            for file in glob(str(self.absoutput) + f"/*{ext}"):
                print("Removing " + file)
                os.remove(file)
 
 

clean_up_last_run() [2/2]

None clean_up_last_run

(

self

)

Removes the executable, libraries, hidden files, and dictionaries from the output directory.

    def clean_up_last_run(self) -> None:
        """Removes the executable, libraries, hidden files, and dictionaries from the output directory."""
 
        exec = Path(which(self.cmds.exec)).name
 
        for file in [exec, ".env", ".condor"]:
            file = self.absoutput / Path(file)
            if file.is_file():
                os.remove(str(file))
 
        for ext in ["so", "pcm"]:
            for file in glob(str(self.absoutput) + f"/*{ext}"):
                print("Removing " + file)
                os.remove(file)
 
 

get_entries() [1/2]

int get_entries	(		self,
			input
	)

Obtain the number of entries from input n-tuple.

    def get_entries(self, input) -> int:
        """Obtain the number of entries from input n-tuple."""
 
        output = check_output(["printEntries"] + [input.strip('\\"')])
        nevents = int(output.decode().splitlines()[0].replace("\n", ""))
        if nevents == 0:
            raise ValueError("Empty input tree.")
        return nevents
 

get_entries() [2/2]

int get_entries	(		self,
			input
	)

Obtain the number of entries from input n-tuple.

    def get_entries(self, input) -> int:
        """Obtain the number of entries from input n-tuple."""
 
        output = check_output(["printEntries"] + [input.strip('\\"')])
        nevents = int(output.decode().splitlines()[0].replace("\n", ""))
        if nevents == 0:
            raise ValueError("Empty input tree.")
        return nevents
 

parse() [1/2]

None parse	(		self,
		List[str]	args
	)

Put input(s), output, and other arguments in dedicated members.

    def parse(self, args: List[str]) -> None:
        """Put input(s), output, and other arguments in dedicated members."""
 
        synopsis = self._synopsis()
        i = synopsis.index("output")
        if i >= len(args):
            raise ValueError("Unable to identify `output` in given arguments")
 
        self.inputs = args[:i]
        self.output = args[i]
        if len(args) > i:
            self.args = args[i + 1 :]
 

parse() [2/2]

None parse	(		self,
		List[str]	args
	)

Put input(s), output, and other arguments in dedicated members.

    def parse(self, args: List[str]) -> None:
        """Put input(s), output, and other arguments in dedicated members."""
 
        synopsis = self._synopsis()
        i = synopsis.index("output")
        if i >= len(args):
            raise ValueError("Unable to identify `output` in given arguments")
 
        self.inputs = args[:i]
        self.output = args[i]
        if len(args) > i:
            self.args = args[i + 1 :]
 

prepare_fire_and_forget() [1/2]

None prepare_fire_and_forget

(

self

)

Copies the executable, libraries, and dictionaries to the output directory.

    def prepare_fire_and_forget(self) -> None:
        """Copies the executable, libraries, and dictionaries to the output directory."""
 
        # copy executables and libraries to working directory
        exec = Path(which(self.cmds.exec)).resolve()
        copy_exec = self.absoutput / exec.name
        if exec != copy_exec:
            copy2(exec, self.output)
            exec = copy_exec
        self.cmds.exec = str(exec)
 
        # get the list of dependencies (for binary executables only)
        mime_type = os.popen(f"file --mime-type {exec}").read()
        cmd = ""
        if sys.platform == "darwin": # i.e. macOS (unfortunate name conflict)
            if not "application/x-mach-binary" in mime_type:
                return
            cmd = 'otool -L ' + str(exec) + ' | awk \'{print $1}\' | grep "\@rpath" | sed \'s/\@rpath\///g\''
        else: # assuming linux
            if not "application/x-executable" in mime_type:
                return
            cmd = 'ldd -r ' + str(exec) + ' | awk \'{print $1}\' | grep "^lib"'
        dependencies = os.popen(cmd).read().splitlines()
 
        # get the list of local libraries
        local_libs = glob("/builds/cms-analysis/general/DasAnalysisSystem/Core/Installer/tools/lib64/*")
 
        # copy the necessary libraries and dictionaries in local working directory
        for dep in dependencies:
            stem = Path(dep).stem
            for ext in ["_rdict.pcm", ".rootmap", ".so", ".dylib"]:
                candidate = stem + ext
                for file in local_libs:
                    if candidate in file and Path(file).is_file():
                        copy2(file, self.absoutput)
 

prepare_fire_and_forget() [2/2]

None prepare_fire_and_forget

(

self

)

Copies the executable, libraries, and dictionaries to the output directory.

    def prepare_fire_and_forget(self) -> None:
        """Copies the executable, libraries, and dictionaries to the output directory."""
 
        # copy executables and libraries to working directory
        exec = Path(which(self.cmds.exec)).resolve()
        copy_exec = self.absoutput / exec.name
        if exec != copy_exec:
            copy2(exec, self.output)
            exec = copy_exec
        self.cmds.exec = str(exec)
 
        # get the list of dependencies (for binary executables only)
        mime_type = os.popen(f"file --mime-type {exec}").read()
        cmd = ""
        if sys.platform == "darwin": # i.e. macOS (unfortunate name conflict)
            if not "application/x-mach-binary" in mime_type:
                return
            cmd = 'otool -L ' + str(exec) + ' | awk \'{print $1}\' | grep "\@rpath" | sed \'s/\@rpath\///g\''
        else: # assuming linux
            if not "application/x-executable" in mime_type:
                return
            cmd = 'ldd -r ' + str(exec) + ' | awk \'{print $1}\' | grep "^lib"'
        dependencies = os.popen(cmd).read().splitlines()
 
        # get the list of local libraries
        local_libs = glob("/builds/cms-analysis/general/DasAnalysisSystem/Core/Installer/tools/lib64/*")
 
        # copy the necessary libraries and dictionaries in local working directory
        for dep in dependencies:
            stem = Path(dep).stem
            for ext in ["_rdict.pcm", ".rootmap", ".so", ".dylib"]:
                candidate = stem + ext
                for file in local_libs:
                    if candidate in file and Path(file).is_file():
                        copy2(file, self.absoutput)
 

prepare_io() [1/2]

None prepare_io	(		self,
		bool	multi = True
	)

Prepare input and output files and directories.

    def prepare_io(self, multi: bool = True) -> None:
        """Prepare input and output files and directories."""
 
        self._prepare_input()
        self.absoutput = Path(self.output).resolve()
 
        if multi:
            self.inputs = [shlex.quote(x) for x in self.inputs]
            self._prepare_multi_output()
        else:
            self._prepare_single_output()
 

prepare_io() [2/2]

None prepare_io	(		self,
		bool	multi = True
	)

Prepare input and output files and directories.

    def prepare_io(self, multi: bool = True) -> None:
        """Prepare input and output files and directories."""
 
        self._prepare_input()
        self.absoutput = Path(self.output).resolve()
 
        if multi:
            self.inputs = [shlex.quote(x) for x in self.inputs]
            self._prepare_multi_output()
        else:
            self._prepare_single_output()
 

Member Data Documentation

absinputs

absinputs

absoutput

absoutput

args

args

cmds

cmds

extension

extension

helper

helper

inputs

inputs

name

name

nSplit

nSplit

output

output

Path

Path

static

splittable

splittable

str

str

static

---

The documentation for this class was generated from the following file:

• /builds/cms-analysis/general/DasAnalysisSystem/Core/Installer/build/Darwin/python/prefix.py