prefix Namespace Reference

Classes
class	PrefixCommand

Functions
def	preparse (List[str] argv, str tutorial, bool multi_opt=False, bool dag_opt=False, bool condor=False)
None	git_hash (str exec)
def	find_institute ()
None	tweak_helper_multi (PrefixCommand prefix, bool multi_opt=True, bool dag_opt=False, bool condor=False)
None	diagnostic (Exception exception)
def	print_slice (formatting, line, nNow, nSplit, end)
def	copy_condor_config (output)

Variables
list	__all__
	NPROC = multiprocessing.cpu_count()

Function Documentation

copy_condor_config()

def copy_condor_config

(

output

)

Copy the HTC config to the running directory.

def copy_condor_config(output):
    """Copy the HTC config to the running directory."""
 
    copy2(Path(__file__).parent / "condor.jdl", str(output) + "/.condor")

diagnostic()

None diagnostic

(

Exception

exception

)

Common diagnostic message for all prefix commands in case of exception.

def diagnostic(exception: Exception) -> None:
    """Common diagnostic message for all prefix commands in case of exception."""
    prefix = os.path.basename(sys.argv[0])
    versions = (
        check_output("printDarwinSoftwareVersion", shell=True)
        .decode()
        .strip()
        .split("\n")
    )
    versions += [f"Python {sys.version_info.major}.{sys.version_info.minor}"]
    print(
        f"\x1B[31m{prefix}: \x1B[1m{exception}\x1B[22m",
        ", ".join(versions),
        "Consider opening a GitLab issue if you don't understand the cause of the failure.\x1B[0m",
        sep="\n",
    )
 
 

find_institute()

def find_institute

(

)

Identifies the host from the name of the machine

def find_institute():
    """Identifies the host from the name of the machine"""
 
    clusters = {
        r"naf-cms[1-2][0-9].desy.de": "DESY",
        r".phy.pku.edu.cn": "PKU",
        r"lxplus[0-9]*.cern.ch": "CERN",
    }
 
    hostname = socket.gethostname()
    for key, value in clusters.items():
        if re.search(key, hostname):
            return value
 
    raise RuntimeError("The host could not be identified")
 
 

git_hash()

None git_hash

(

str

exec

)

Returns the SHA of the executable at compile time.

def git_hash(exec: str) -> None:
    """Returns the SHA of the executable at compile time."""
    print(check_output(exec + " -g", shell=True).decode(), end="")
 
 

preparse()

def preparse	(	List[str]	argv,
		str	tutorial,
		bool	multi_opt = False,
		bool	dag_opt = False,
		bool	condor = False
	)

Parser for the prefix command itself, relying on ArgumentParser.

def preparse(
    argv: List[str],
    tutorial: str,
    multi_opt: bool = False,
    dag_opt: bool = False,
    condor: bool = False,
):
    """Parser for the prefix command itself, relying on ArgumentParser."""
 
    parser = ArgumentParser(add_help=False)
 
    parser.add_argument("exec", nargs="?")
    if multi_opt:
        parser.add_argument("-b", "--background", action="store_true")
        parser.add_argument("-d", "--dry-run", action="store_true")
    if dag_opt:
        parser.add_argument("-D", "--dag", type=Path, nargs='?', default="dag")
    if condor:
        parser.add_argument("-n", "--memory-needs", type=int, default=1024)
 
    # helper
    parser.add_argument("-h", "--help", action="store_true")
    parser.add_argument("-t", "--tutorial", action="store_true")
    parser.add_argument("-g", "--git", action="store_true")
    if not dag_opt:
        parser.add_argument("-e", "--example", action="store_true")
 
    # common
    parser.add_argument("-v", "--verbose", action="store_true")
    parser.add_argument("-m", "--mute", action="store_true")
    parser.add_argument("-f", "--fill", action="store_true")
    parser.add_argument("-F", "--Friend", action="store_true")
    parser.add_argument("-s", "--syst", action="store_true")
    parser.add_argument("-c", "--config", type=Path)
    parser.add_argument("-j", "--nSplit", type=int, default=NPROC)
 
    # custom will be in args
    cmds, args = parser.parse_known_intermixed_args(argv[1:])
 
    if any(x in args for x in ["-k", "--nNow"]):
        raise ValueError("Giving `-k` or `--nNow` is meaningless in this context.")
    unknown_explicit_args = [arg for arg in args if arg[0] == "-"]
    if len(unknown_explicit_args) > 0:
        raise ValueError(
            f"No unknown explicit option is allowed: {unknown_explicit_args}"
        )
 
    # restore explicit options
    if cmds.verbose:
        args += ["-v"]
    if cmds.mute:
        args += ["-m"]
    if cmds.fill:
        args += ["-f"]
    if cmds.Friend:
        args += ["-F"]
    if cmds.syst:
        args += ["-s"]
    if cmds.config:
        args += ["-c", str(cmds.config.resolve())]
 
    if not cmds.exec:
        if cmds.help or len(argv) == 1:
            prefix = os.path.basename(sys.argv[0])
            print(
                f"\33[1m{prefix} exec args [...]\33[0m",
                "where\texec = a command using `Darwin::Tools::Options` with `Darwin::Tools::split`",
                "\targs = the arguments of the same command, one being called exactly `output`",
                sep="\n",
            )
        if cmds.tutorial:
            # boost::program_options::options_description::m_default_line_length = 80
            lines = textwrap.TextWrapper(width=80).wrap(text=tutorial)
            print(*lines, sep="\n")
        parser.exit()
 
    return cmds, args
 
 

print_slice()

def print_slice	(		formatting,
			line,
			nNow,
			nSplit,
			end
	)

Prefix the output stream with the slice. The formatting of the
printout is preserved but the slice is printed in default formatting.

def print_slice(formatting, line, nNow, nSplit, end):
    """
    Prefix the output stream with the slice. The formatting of the
    printout is preserved but the slice is printed in default formatting.
    """
 
    RESET = "\x1B[0m"
 
    # find the special characters to encode the colour and the highlighting
    special_chars = re.findall(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])", line)
 
    # if any was found, check if it should be reset to default
    if special_chars and special_chars[-1] == RESET:
        formatting = RESET
    else:
        formatting += "".join(special_chars)
 
    print(f"{nNow}/{nSplit}\t{formatting}{line}{RESET}", end=end)
 
    return formatting
 
 

tweak_helper_multi()

None tweak_helper_multi	(	PrefixCommand	prefix,
		bool	multi_opt = True,
		bool	dag_opt = False,
		bool	condor = False
	)

Tweak helper to use a directory as output rather than a file
and to reflect the number of cores of the machine.

def tweak_helper_multi(
    prefix: PrefixCommand,
    multi_opt: bool = True,
    dag_opt: bool = False,
    condor: bool = False,
) -> None:
    """Tweak helper to use a directory as output rather than a file
    and to reflect the number of cores of the machine."""
 
    print(f"\33[1m{prefix.name} \33[0m", end="")
    for row in prefix.helper:
        if "output" in row:
            row = row.replace("ROOT file", "directory")
        if "nNow" in row:
            continue
        if "nSplit" in row:
            row = row.replace(
                "(=1)" if NPROC < 10 else "(=1) ", "(=" + str(NPROC) + ")"
            )
        print(row)
        if "Helper" in row:
            if multi_opt:
                print("  -b [ --background ]       Do not wait for job to finish")
                print("  -d [ --dry-run ]          Run until the actual executation")
            if dag_opt:
                print("  -D [ --dag ]              Indicate DAG file")
            if condor:
                print("  -n [ --memory-needs] arg  Memory needs for HTCondor job")
    print("")
 
 

Variable Documentation

__all__

list __all__

private

Initial value:

=  [
    "PrefixCommand",
    "preparse",
    "tweak_helper_multi",
    "diagnostic",
    "git_hash",
    "copy_condor_config",
    "find_institute",
]

NPROC

NPROC = multiprocessing.cpu_count()