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_single (self)
None	prepare_multi (self)
None	prepare_fire_and_forget (self)
None	__init__ (self, str prefix, List[str] cmds)
None	parse (self, List[str] args)
int	get_entries (self, input)
None	prepare_single (self)
None	prepare_multi (self)
None	prepare_fire_and_forget (self)

Public Attributes
	name
	cmds
	helper
	extension
	inputs
	output
	args
	absoutput
	absinputs

Static Public Attributes
	str
	Path

Private Member Functions
str	_help_text (self)
List[str]	_synopsis (self)
str	_help_text (self)
List[str]	_synopsis (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 not can_parallelize(self.helper[1:]):
            raise ValueError(
                f"`{self.cmds.exec}`"
                + " does not match the requirements to be run with this prefix command: "
                + "exactly one `output` argument and `nSplit` argument must be defined."
            )
 
        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 not can_parallelize(self.helper[1:]):
            raise ValueError(
                f"`{self.cmds.exec}`"
                + " does not match the requirements to be run with this prefix command: "
                + "exactly one `output` argument and `nSplit` argument must be defined."
            )
 
        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
 

_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:]
 

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)).absolute()
        copy_exec = self.absoutput / exec.name
        if exec != copy_exec:
            copy2(exec, self.output)
            exec = copy_exec
        self.cmds.exec = str(exec)
 
        for lib in glob(os.environ["DARWIN_FIRE_AND_FORGET"] + "/*so"):
            copy2(lib, self.output)
        for dic in glob(os.environ["DARWIN_FIRE_AND_FORGET"] + "/*pcm"):
            copy2(dic, self.output)
 
 

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)).absolute()
        copy_exec = self.absoutput / exec.name
        if exec != copy_exec:
            copy2(exec, self.output)
            exec = copy_exec
        self.cmds.exec = str(exec)
 
        for lib in glob(os.environ["DARWIN_FIRE_AND_FORGET"] + "/*so"):
            copy2(lib, self.output)
        for dic in glob(os.environ["DARWIN_FIRE_AND_FORGET"] + "/*pcm"):
            copy2(dic, self.output)
 
 

prepare_multi() [1/2]

None prepare_multi

(

self

)

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

    def prepare_multi(self) -> None:
        """Prepare the members for parallel running (local or on farm)."""
 
        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).absolute()) for x in self.absinputs]
 
        self.inputs = ['"' + x + '"' for x in self.inputs]
        self.absoutput = Path(self.output).absolute()
        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.cmds.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)
 
        # the following lines are specific for EOS at CERN, where
        # `/eos/home-b/bejones` should be replaced with `/eos/user/b/bejones`
        try:
            institute = find_institute()
            if find_institute() == "CERN" and self.absoutput.parts[1] == "eos":
                user = str(self.absoutput.parts[3])
                homedir = Path(*self.absoutput.parts[3:])
                self.absoutput = Path(f"/eos/user/{user[0]}") / homedir
        except RuntimeError:
            pass
 

prepare_multi() [2/2]

None prepare_multi

(

self

)

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

    def prepare_multi(self) -> None:
        """Prepare the members for parallel running (local or on farm)."""
 
        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).absolute()) for x in self.absinputs]
 
        self.inputs = ['"' + x + '"' for x in self.inputs]
        self.absoutput = Path(self.output).absolute()
        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.cmds.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)
 
        # the following lines are specific for EOS at CERN, where
        # `/eos/home-b/bejones` should be replaced with `/eos/user/b/bejones`
        try:
            institute = find_institute()
            if find_institute() == "CERN" and self.absoutput.parts[1] == "eos":
                user = str(self.absoutput.parts[3])
                homedir = Path(*self.absoutput.parts[3:])
                self.absoutput = Path(f"/eos/user/{user[0]}") / homedir
        except RuntimeError:
            pass
 

prepare_single() [1/2]

None prepare_single

(

self

)

Prepare the working area for a single-run command.

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

prepare_single() [2/2]

None prepare_single

(

self

)

Prepare the working area for a single-run command.

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

Member Data Documentation

absinputs

absinputs

absoutput

absoutput

args

args

cmds

cmds

extension

extension

helper

helper

inputs

inputs

name

name

output

output

Path

Path

static

str

str

static

---

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

• /builds/Proto/Darwin/build/python/prefix.py