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)
bool	can_parallelize (helper)
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	copy_condor_config (str output)

Variables
list	__all__
	NPROC = multiprocessing.cpu_count()

Function Documentation

can_parallelize()

bool can_parallelize

(

helper

)

Checks if the command can be parallelized.

def can_parallelize(helper) -> bool:
    """Checks if the command can be parallelized."""
    nSplit_exists = ["nSplit" in row for row in helper].count(True)
    output_exists = ["output" in row for row in helper].count(True)
    return nSplit_exists and output_exists
 
 

copy_condor_config()

def copy_condor_config

(

str

output

)

Identifies the proper HTC config for the local host.

def copy_condor_config(output: str):
    """Identifies the proper HTC config for the local host."""
 
    institute = ""
    try:
        institute = find_institute()
    except RuntimeError as e:
        print(e)
        institute = "default"
 
    copy2(Path(__file__).parent / f"{institute}.jdl", output + "/.condor")
    copy2(Path(__file__).parent / ".run", output)

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)
    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.absolute())]
 
    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
 
 

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 DAGMan directory location")
            if condor:
                try:
                    institute = find_institute()
                    if institute in ["DESY", "CERN"]:
                        print(
                            "  -n [ --memory-needs] arg  Memory needs for HTCondor job"
                        )
                except RuntimeError:
                    pass
    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()